Merge pull request #9972 from sreecha/cq_create_api_changes

Completion Queue creation API changes (C-Core and C++ only)

Makefile

@@ -419,7 +419,7 @@ E = @echo
 Q = @
 endif
 
-CORE_VERSION = 3.0.0-dev
+CORE_VERSION = 4.0.0-dev
 CPP_VERSION = 1.4.0-dev
 CSHARP_VERSION = 1.4.0-dev
 
@@ -469,7 +469,7 @@ SHARED_EXT_CORE = dll
 SHARED_EXT_CPP = dll
 SHARED_EXT_CSHARP = dll
 SHARED_PREFIX =
-SHARED_VERSION_CORE = -3
+SHARED_VERSION_CORE = -4
 SHARED_VERSION_CPP = -1
 SHARED_VERSION_CSHARP = -1
 else ifeq ($(SYSTEM),Darwin)
@@ -2554,7 +2554,7 @@ install-shared_c: shared_c strip-shared_c install-pkg-config_c
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE)-dll.a $(prefix)/lib/libgpr.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgpr.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgpr.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgpr.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE)"
@@ -2563,7 +2563,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE)-dll.a $(prefix)/lib/libgrpc.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE)"
@@ -2572,7 +2572,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE)-dll.a $(prefix)/lib/libgrpc_cronet.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc_cronet.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc_cronet.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc_cronet.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE)"
@@ -2581,7 +2581,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE)-dll.a $(prefix)/lib/libgrpc_unsecure.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc_unsecure.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc_unsecure.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(prefix)/lib/libgrpc_unsecure.so
 endif
 ifneq ($(SYSTEM),MINGW32)
@@ -2598,7 +2598,7 @@ install-shared_cxx: shared_cxx strip-shared_cxx install-shared_c install-pkg-con
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc++$(SHARED_VERSION_CPP)-dll.a $(prefix)/lib/libgrpc++.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc++$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc++$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc++$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc++_cronet$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP)"
@@ -2607,7 +2607,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc++_cronet$(SHARED_VERSION_CPP)-dll.a $(prefix)/lib/libgrpc++_cronet.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc++_cronet$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_cronet.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc++_cronet$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_cronet.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc++_cronet$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_cronet.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc++_error_details$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP)"
@@ -2616,7 +2616,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc++_error_details$(SHARED_VERSION_CPP)-dll.a $(prefix)/lib/libgrpc++_error_details.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc++_error_details$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_error_details.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc++_error_details$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_error_details.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc++_error_details$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_error_details.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc++_reflection$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP)"
@@ -2625,7 +2625,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc++_reflection$(SHARED_VERSION_CPP)-dll.a $(prefix)/lib/libgrpc++_reflection.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc++_reflection$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_reflection.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc++_reflection$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_reflection.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc++_reflection$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_reflection.so
 endif
 	$(E) "[INSTALL] Installing $(SHARED_PREFIX)grpc++_unsecure$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP)"
@@ -2634,7 +2634,7 @@ endif
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc++_unsecure$(SHARED_VERSION_CPP)-dll.a $(prefix)/lib/libgrpc++_unsecure.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc++_unsecure$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_unsecure.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc++_unsecure$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_unsecure.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc++_unsecure$(SHARED_VERSION_CPP).$(SHARED_EXT_CPP) $(prefix)/lib/libgrpc++_unsecure.so
 endif
 ifneq ($(SYSTEM),MINGW32)
@@ -2651,7 +2651,7 @@ install-shared_csharp: shared_csharp strip-shared_csharp
 ifeq ($(SYSTEM),MINGW32)
 	$(Q) $(INSTALL) $(LIBDIR)/$(CONFIG)/libgrpc_csharp_ext$(SHARED_VERSION_CSHARP)-dll.a $(prefix)/lib/libgrpc_csharp_ext.a
 else ifneq ($(SYSTEM),Darwin)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc_csharp_ext$(SHARED_VERSION_CSHARP).$(SHARED_EXT_CSHARP) $(prefix)/lib/libgrpc_csharp_ext.so.3
+	$(Q) ln -sf $(SHARED_PREFIX)grpc_csharp_ext$(SHARED_VERSION_CSHARP).$(SHARED_EXT_CSHARP) $(prefix)/lib/libgrpc_csharp_ext.so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc_csharp_ext$(SHARED_VERSION_CSHARP).$(SHARED_EXT_CSHARP) $(prefix)/lib/libgrpc_csharp_ext.so
 endif
 ifneq ($(SYSTEM),MINGW32)
@@ -2816,8 +2816,8 @@ $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE): $(LIBGPR_OB
 ifeq ($(SYSTEM),Darwin)
 	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -install_name $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) -dynamiclib -o $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGPR_OBJS) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
 else
-	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgpr.so.3 -o $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGPR_OBJS) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
-	$(Q) ln -sf $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).so.3
+	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgpr.so.4 -o $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGPR_OBJS) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
+	$(Q) ln -sf $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).so.4
 	$(Q) ln -sf $(SHARED_PREFIX)gpr$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgpr$(SHARED_VERSION_CORE).so
 endif
 endif
@@ -3160,8 +3160,8 @@ $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE): $(LIBGRPC_
 ifeq ($(SYSTEM),Darwin)
 	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -install_name $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) -dynamiclib -o $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(OPENSSL_MERGE_LIBS) $(LDLIBS_SECURE) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
 else
-	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgrpc.so.3 -o $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(OPENSSL_MERGE_LIBS) $(LDLIBS_SECURE) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).so.3
+	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgrpc.so.4 -o $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(OPENSSL_MERGE_LIBS) $(LDLIBS_SECURE) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
+	$(Q) ln -sf $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc$(SHARED_VERSION_CORE).so
 endif
 endif
@@ -3450,8 +3450,8 @@ $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE): $(L
 ifeq ($(SYSTEM),Darwin)
 	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -install_name $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) -dynamiclib -o $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_CRONET_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(OPENSSL_MERGE_LIBS) $(LDLIBS_SECURE) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
 else
-	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgrpc_cronet.so.3 -o $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_CRONET_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(OPENSSL_MERGE_LIBS) $(LDLIBS_SECURE) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).so.3
+	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgrpc_cronet.so.4 -o $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_CRONET_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(OPENSSL_MERGE_LIBS) $(LDLIBS_SECURE) $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
+	$(Q) ln -sf $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc_cronet$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc_cronet$(SHARED_VERSION_CORE).so
 endif
 endif
@@ -3979,8 +3979,8 @@ $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE): $
 ifeq ($(SYSTEM),Darwin)
 	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -install_name $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) -dynamiclib -o $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_UNSECURE_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
 else
-	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgrpc_unsecure.so.3 -o $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_UNSECURE_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
-	$(Q) ln -sf $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).so.3
+	$(Q) $(LD) $(LDFLAGS) -L$(LIBDIR)/$(CONFIG) -shared -Wl,-soname,libgrpc_unsecure.so.4 -o $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBGRPC_UNSECURE_OBJS) $(LIBDIR)/$(CONFIG)/libgpr.a $(ZLIB_MERGE_LIBS) $(CARES_MERGE_LIBS) $(LDLIBS)
+	$(Q) ln -sf $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).so.4
 	$(Q) ln -sf $(SHARED_PREFIX)grpc_unsecure$(SHARED_VERSION_CORE).$(SHARED_EXT_CORE) $(LIBDIR)/$(CONFIG)/libgrpc_unsecure$(SHARED_VERSION_CORE).so
 endif
 endif

build.yaml

@@ -12,7 +12,7 @@ settings:
   '#08': Use "-preN" suffixes to identify pre-release versions
   '#09': Per-language overrides are possible with (eg) ruby_version tag here
   '#10': See the expand_version.py for all the quirks here
-  core_version: 3.0.0-dev
+  core_version: 4.0.0-dev
   g_stands_for: gentle
   version: 1.4.0-dev
 filegroups:

build_config.rb

@@ -28,5 +28,5 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 module GrpcBuildConfig
-  CORE_WINDOWS_DLL = '/tmp/libs/opt/grpc-3.dll'
+  CORE_WINDOWS_DLL = '/tmp/libs/opt/grpc-4.dll'
 end

include/grpc++/impl/codegen/client_unary_call.h

@@ -52,7 +52,7 @@ template <class InputMessage, class OutputMessage>
 Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
                          ClientContext* context, const InputMessage& request,
                          OutputMessage* result) {
-  CompletionQueue cq;
+  CompletionQueue cq(true);  // Pluckable completion queue
   Call call(channel->CreateCall(method, context, &cq));
   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
             CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,

include/grpc++/impl/codegen/completion_queue.h

@@ -102,10 +102,7 @@ class CompletionQueue : private GrpcLibraryCodegen {
  public:
   /// Default constructor. Implicitly creates a \a grpc_completion_queue
   /// instance.
-  CompletionQueue() {
-    cq_ = g_core_codegen_interface->grpc_completion_queue_create(nullptr);
-    InitialAvalanching();  // reserve this for the future shutdown
-  }
+  CompletionQueue() : CompletionQueue(false) {}
 
   /// Wrap \a take, taking ownership of the instance.
   ///
@@ -218,6 +215,18 @@ class CompletionQueue : private GrpcLibraryCodegen {
                                   const InputMessage& request,
                                   OutputMessage* result);
 
+  /// Private constructor of CompletionQueue only visible to friend classes
+  CompletionQueue(bool is_pluck) {
+    if (is_pluck) {
+      cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_pluck(
+          nullptr);
+    } else {
+      cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_next(
+          nullptr);
+    }
+    InitialAvalanching();  // reserve this for the future shutdown
+  }
+
   NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
 
   /// Wraps \a grpc_completion_queue_pluck.
@@ -237,6 +246,12 @@ class CompletionQueue : private GrpcLibraryCodegen {
 
   /// Performs a single polling pluck on \a tag.
   /// \warning Must not be mixed with calls to \a Next.
+  ///
+  /// TODO: sreek - This calls tag->FinalizeResult() even if the cq_ is already
+  /// shutdown. This is most likely a bug and if it is a bug, then change this
+  /// implementation to simple call the other TryPluck function with a zero
+  /// timeout. i.e:
+  ///      TryPluck(tag, gpr_time_0(GPR_CLOCK_REALTIME))
   void TryPluck(CompletionQueueTag* tag) {
     auto deadline = g_core_codegen_interface->gpr_time_0(GPR_CLOCK_REALTIME);
     auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
@@ -248,6 +263,23 @@ class CompletionQueue : private GrpcLibraryCodegen {
     GPR_CODEGEN_ASSERT(!tag->FinalizeResult(&ignored, &ok));
   }
 
+  /// Performs a single polling pluck on \a tag. Calls tag->FinalizeResult if
+  /// the pluck() was successful and returned the tag.
+  ///
+  /// This exects tag->FinalizeResult (if called) to return 'false' i.e expects
+  /// that the tag is internal not something that is returned to the user.
+  void TryPluck(CompletionQueueTag* tag, gpr_timespec deadline) {
+    auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
+        cq_, tag, deadline, nullptr);
+    if (ev.type == GRPC_QUEUE_TIMEOUT || ev.type == GRPC_QUEUE_SHUTDOWN) {
+      return;
+    }
+
+    bool ok = ev.success != 0;
+    void* ignored = tag;
+    GPR_CODEGEN_ASSERT(!tag->FinalizeResult(&ignored, &ok));
+  }
+
   grpc_completion_queue* cq_;  // owned
 
   gpr_atm avalanches_in_flight_;

include/grpc++/impl/codegen/core_codegen.h

@@ -38,6 +38,7 @@
 
 #include <grpc++/impl/codegen/core_codegen_interface.h>
 #include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
 #include <grpc/impl/codegen/grpc_types.h>
 
 namespace grpc {
@@ -45,7 +46,10 @@ namespace grpc {
 /// Implementation of the core codegen interface.
 class CoreCodegen : public CoreCodegenInterface {
  private:
-  grpc_completion_queue* grpc_completion_queue_create(void* reserved) override;
+  grpc_completion_queue* grpc_completion_queue_create_for_next(
+      void* reserved) override;
+  grpc_completion_queue* grpc_completion_queue_create_for_pluck(
+      void* reserved) override;
   void grpc_completion_queue_destroy(grpc_completion_queue* cq) override;
   grpc_event grpc_completion_queue_pluck(grpc_completion_queue* cq, void* tag,
                                          gpr_timespec deadline,

include/grpc++/impl/codegen/core_codegen_interface.h

@@ -59,7 +59,9 @@ class CoreCodegenInterface {
   virtual void assert_fail(const char* failed_assertion, const char* file,
                            int line) = 0;
 
-  virtual grpc_completion_queue* grpc_completion_queue_create(
+  virtual grpc_completion_queue* grpc_completion_queue_create_for_next(
+      void* reserved) = 0;
+  virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck(
       void* reserved) = 0;
   virtual void grpc_completion_queue_destroy(grpc_completion_queue* cq) = 0;
   virtual grpc_event grpc_completion_queue_pluck(grpc_completion_queue* cq,

include/grpc++/impl/codegen/server_context.h

@@ -40,6 +40,7 @@
 
 #include <grpc/impl/codegen/compression_types.h>
 
+#include <grpc++/impl/codegen/completion_queue_tag.h>
 #include <grpc++/impl/codegen/config.h>
 #include <grpc++/impl/codegen/create_auth_context.h>
 #include <grpc++/impl/codegen/metadata_map.h>
@@ -211,6 +212,8 @@ class ServerContext {
   class CompletionOp;
 
   void BeginCompletionOp(Call* call);
+  // Return the tag queued by BeginCompletionOp()
+  CompletionQueueTag* GetCompletionOpTag();
 
   ServerContext(gpr_timespec deadline, grpc_metadata_array* arr);
 

include/grpc++/impl/codegen/sync_stream.h

@@ -155,7 +155,9 @@ class ClientReader final : public ClientReaderInterface<R> {
   template <class W>
   ClientReader(ChannelInterface* channel, const RpcMethod& method,
                ClientContext* context, const W& request)
-      : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+      : context_(context),
+        cq_(true),  // Pluckable cq
+        call_(channel->CreateCall(method, context, &cq_)) {
     CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
               CallOpClientSendClose>
         ops;
@@ -227,7 +229,9 @@ class ClientWriter : public ClientWriterInterface<W> {
   template <class R>
   ClientWriter(ChannelInterface* channel, const RpcMethod& method,
                ClientContext* context, R* response)
-      : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+      : context_(context),
+        cq_(true),  // Pluckable cq
+        call_(channel->CreateCall(method, context, &cq_)) {
     finish_ops_.RecvMessage(response);
     finish_ops_.AllowNoMessage();
 
@@ -325,7 +329,9 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
   /// Blocking create a stream.
   ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method,
                      ClientContext* context)
-      : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+      : context_(context),
+        cq_(true),  // Pluckable cq
+        call_(channel->CreateCall(method, context, &cq_)) {
     if (!context_->initial_metadata_corked_) {
       CallOpSet<CallOpSendInitialMetadata> ops;
       ops.SendInitialMetadata(context->send_initial_metadata_,
@@ -562,7 +568,7 @@ class ServerReaderWriterBody final {
   Call* const call_;
   ServerContext* const ctx_;
 };
-}
+}  // namespace internal
 
 // class to represent the user API for a bidirectional streaming call
 template <class W, class R>

include/grpc/grpc.h

@@ -95,10 +95,10 @@ GRPCAPI const char *grpc_g_stands_for(void);
 
 /** Specifies the type of APIs to use to pop events from the completion queue */
 typedef enum {
-  /* Events are popped out by calling grpc_completion_queue_next() API ONLY */
+  /** Events are popped out by calling grpc_completion_queue_next() API ONLY */
   GRPC_CQ_NEXT = 1,
 
-  /* Events are popped out by calling grpc_completion_queue_pluck() API ONLY */
+  /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
   GRPC_CQ_PLUCK
 } grpc_cq_completion_type;
 
@@ -116,15 +116,15 @@ typedef enum {
       restriction on the type of file descriptors the pollset may contain */
   GRPC_CQ_DEFAULT_POLLING,
 
-  /* Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will
-     not contain any 'listening file descriptors' (i.e file descriptors used to
-     listen to incoming channels) */
+  /** Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will
+      not contain any 'listening file descriptors' (i.e file descriptors used to
+      listen to incoming channels) */
   GRPC_CQ_NON_LISTENING,
 
-  /* The completion queue will not have an associated pollset. Note that
-     grpc_completion_queue_next() or grpc_completion_queue_pluck() MUST still be
-     called to pop events from the completion queue; it is not required to call
-     them actively to make I/O progress */
+  /** The completion queue will not have an associated pollset. Note that
+      grpc_completion_queue_next() or grpc_completion_queue_pluck() MUST still
+      be called to pop events from the completion queue; it is not required to
+      call them actively to make I/O progress */
   GRPC_CQ_NON_POLLING
 } grpc_cq_polling_type;
 
@@ -159,7 +159,9 @@ GRPCAPI grpc_completion_queue *grpc_completion_queue_create_for_pluck(
     void *reserved);
 
 /** Create a completion queue */
-GRPCAPI grpc_completion_queue *grpc_completion_queue_create(void *reserved);
+GRPCAPI grpc_completion_queue *grpc_completion_queue_create(
+    const grpc_completion_queue_factory *factory,
+    const grpc_completion_queue_attributes *attributes, void *reserved);
 
 /** Blocks until an event is available, the completion queue is being shut down,
     or deadline is reached.

include/grpc/impl/codegen/grpc_types.h

@@ -401,8 +401,11 @@ typedef enum grpc_completion_type {
 typedef struct grpc_event {
   /** The type of the completion. */
   grpc_completion_type type;
-  /** non-zero if the operation was successful, 0 upon failure.
-      Only GRPC_OP_COMPLETE can succeed or fail. */
+  /** If the grpc_completion_type is GRPC_OP_COMPLETE, this field indicates
+      whether the operation was successful or not; 0 in case of failure and
+      non-zero in case of success.
+      If grpc_completion_type is GRPC_QUEUE_SHUTDOWN or GRPC_QUEUE_TIMEOUT, this
+      field is guaranteed to be 0 */
   int success;
   /** The tag passed to grpc_call_start_batch etc to start this operation.
       Only GRPC_OP_COMPLETE has a tag. */

src/core/lib/surface/completion_queue.c

@@ -64,6 +64,10 @@ typedef struct {
 struct grpc_completion_queue {
   /** owned by pollset */
   gpr_mu *mu;
+
+  grpc_cq_completion_type completion_type;
+  grpc_cq_polling_type polling_type;
+
   /** completed events */
   grpc_cq_completion completed_head;
   grpc_cq_completion *completed_tail;
@@ -79,6 +83,7 @@ struct grpc_completion_queue {
   int shutdown_called;
   int is_server_cq;
   /** Can the server cq accept incoming channels */
+  /* TODO: sreek - This will no longer be needed. Use polling_type set */
   int is_non_listening_server_cq;
   int num_pluckers;
   plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS];
@@ -110,13 +115,17 @@ int grpc_cq_event_timeout_trace;
 static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
                                      grpc_error *error);
 
-grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
+grpc_completion_queue *grpc_completion_queue_create_internal(
+    grpc_cq_completion_type completion_type,
+    grpc_cq_polling_type polling_type) {
   grpc_completion_queue *cc;
-  GPR_ASSERT(!reserved);
 
-  GPR_TIMER_BEGIN("grpc_completion_queue_create", 0);
+  GPR_TIMER_BEGIN("grpc_completion_queue_create_internal", 0);
 
-  GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved));
+  GRPC_API_TRACE(
+      "grpc_completion_queue_create_internal(completion_type=%d, "
+      "polling_type=%d)",
+      2, (completion_type, polling_type));
 
   cc = gpr_zalloc(sizeof(grpc_completion_queue) + grpc_pollset_size());
   grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu);
@@ -125,6 +134,9 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
   cc->outstanding_tag_capacity = 0;
 #endif
 
+  cc->completion_type = completion_type;
+  cc->polling_type = polling_type;
+
   /* Initial ref is dropped by grpc_completion_queue_shutdown */
   gpr_ref_init(&cc->pending_events, 1);
   /* One for destroy(), one for pollset_shutdown */
@@ -143,11 +155,19 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
   grpc_closure_init(&cc->pollset_shutdown_done, on_pollset_shutdown_done, cc,
                     grpc_schedule_on_exec_ctx);
 
-  GPR_TIMER_END("grpc_completion_queue_create", 0);
+  GPR_TIMER_END("grpc_completion_queue_create_internal", 0);
 
   return cc;
 }
 
+grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc) {
+  return cc->completion_type;
+}
+
+grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc) {
+  return cc->polling_type;
+}
+
 #ifdef GRPC_CQ_REF_COUNT_DEBUG
 void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason,
                           const char *file, int line) {
@@ -347,6 +367,13 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
   grpc_event ret;
   gpr_timespec now;
 
+  if (cc->completion_type != GRPC_CQ_NEXT) {
+    gpr_log(GPR_ERROR,
+            "grpc_completion_queue_next() cannot be called on this completion "
+            "queue since its completion type is not GRPC_CQ_NEXT");
+    abort();
+  }
+
   GPR_TIMER_BEGIN("grpc_completion_queue_next", 0);
 
   GRPC_API_TRACE(
@@ -516,6 +543,13 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
 
   GPR_TIMER_BEGIN("grpc_completion_queue_pluck", 0);
 
+  if (cc->completion_type != GRPC_CQ_PLUCK) {
+    gpr_log(GPR_ERROR,
+            "grpc_completion_queue_pluck() cannot be called on this completion "
+            "queue since its completion type is not GRPC_CQ_PLUCK");
+    abort();
+  }
+
   if (grpc_cq_pluck_trace) {
     GRPC_API_TRACE(
         "grpc_completion_queue_pluck("
@@ -680,10 +714,14 @@ grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps) {
 }
 
 void grpc_cq_mark_non_listening_server_cq(grpc_completion_queue *cc) {
+  /* TODO: sreek - use cc->polling_type field here and add a validation check
+     (i.e grpc_cq_mark_non_listening_server_cq can only be called on a cc whose
+     polling_type is set to GRPC_CQ_NON_LISTENING */
   cc->is_non_listening_server_cq = 1;
 }
 
 bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc) {
+  /* TODO (sreek) - return (cc->polling_type == GRPC_CQ_NON_LISTENING) */
   return (cc->is_non_listening_server_cq == 1);
 }
 

src/core/lib/surface/completion_queue.h

@@ -99,4 +99,10 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
 void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
 int grpc_cq_is_server_cq(grpc_completion_queue *cc);
 
+grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc);
+grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc);
+
+grpc_completion_queue *grpc_completion_queue_create_internal(
+    grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type);
+
 #endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */

src/core/lib/surface/completion_queue_factory.c

@@ -36,12 +36,15 @@
 
 #include <grpc/support/log.h>
 
-/* TODO (sreek) - Currently this does not use the attributes arg. This will be
-   added in a future PR */
+/*
+ * == Default completion queue factory implementation ==
+ */
+
 static grpc_completion_queue* default_create(
     const grpc_completion_queue_factory* factory,
-    const grpc_completion_queue_attributes* attributes) {
-  return grpc_completion_queue_create(NULL);
+    const grpc_completion_queue_attributes* attr) {
+  return grpc_completion_queue_create_internal(attr->cq_completion_type,
+                                               attr->cq_polling_type);
 }
 
 static grpc_completion_queue_factory_vtable default_vtable = {default_create};
@@ -49,19 +52,24 @@ static grpc_completion_queue_factory_vtable default_vtable = {default_create};
 static const grpc_completion_queue_factory g_default_cq_factory = {
     "Default Factory", NULL, &default_vtable};
 
+/*
+ * == Completion queue factory APIs
+ */
+
 const grpc_completion_queue_factory* grpc_completion_queue_factory_lookup(
     const grpc_completion_queue_attributes* attributes) {
-  /* As we add more fields to grpc_completion_queue_attributes, we may have to
-     change this assert to:
-         GPR_ASSERT (attributes->version >= 1 &&
-             attributes->version <= GRPC_CQ_CURRENT_VERSION) */
-  GPR_ASSERT(attributes->version == 1);
+  GPR_ASSERT(attributes->version >= 1 &&
+             attributes->version <= GRPC_CQ_CURRENT_VERSION);
 
   /* The default factory can handle version 1 of the attributes structure. We
      may have to change this as more fields are added to the structure */
   return &g_default_cq_factory;
 }
 
+/*
+ * == Completion queue creation APIs ==
+ */
+
 grpc_completion_queue* grpc_completion_queue_create_for_next(void* reserved) {
   GPR_ASSERT(!reserved);
   grpc_completion_queue_attributes attr = {1, GRPC_CQ_NEXT,
@@ -75,3 +83,10 @@ grpc_completion_queue* grpc_completion_queue_create_for_pluck(void* reserved) {
                                            GRPC_CQ_DEFAULT_POLLING};
   return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr);
 }
+
+grpc_completion_queue* grpc_completion_queue_create(
+    const grpc_completion_queue_factory* factory,
+    const grpc_completion_queue_attributes* attr, void* reserved) {
+  GPR_ASSERT(!reserved);
+  return factory->vtable->create(factory, attr);
+}

src/core/lib/surface/server.c

@@ -1000,6 +1000,15 @@ void grpc_server_register_completion_queue(grpc_server *server,
   GRPC_API_TRACE(
       "grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
       (server, cq, reserved));
+
+  if (grpc_get_cq_completion_type(cq) != GRPC_CQ_NEXT) {
+    gpr_log(GPR_INFO,
+            "Completion queue which is not of type GRPC_CQ_NEXT is being "
+            "registered as a server-completion-queue");
+    /* Ideally we should log an error and abort but ruby-wrapped-language API
+       calls grpc_completion_queue_pluck() on server completion queues */
+  }
+
   register_completion_queue(server, cq, false, reserved);
 }
 

src/core/lib/surface/version.c

@@ -36,6 +36,6 @@
 
 #include <grpc/grpc.h>
 
-const char *grpc_version_string(void) { return "3.0.0-dev"; }
+const char *grpc_version_string(void) { return "4.0.0-dev"; }
 
 const char *grpc_g_stands_for(void) { return "gentle"; }

src/cpp/common/core_codegen.cc

@@ -54,9 +54,14 @@ struct grpc_byte_buffer;
 
 namespace grpc {
 
-grpc_completion_queue* CoreCodegen::grpc_completion_queue_create(
+grpc_completion_queue* CoreCodegen::grpc_completion_queue_create_for_next(
     void* reserved) {
-  return ::grpc_completion_queue_create(reserved);
+  return ::grpc_completion_queue_create_for_next(reserved);
+}
+
+grpc_completion_queue* CoreCodegen::grpc_completion_queue_create_for_pluck(
+    void* reserved) {
+  return ::grpc_completion_queue_create_for_pluck(reserved);
 }
 
 void CoreCodegen::grpc_completion_queue_destroy(grpc_completion_queue* cq) {

src/cpp/server/server_cc.cc

@@ -124,6 +124,14 @@ class ShutdownTag : public CompletionQueueTag {
   bool FinalizeResult(void** tag, bool* status) { return false; }
 };
 
+class DummyTag : public CompletionQueueTag {
+ public:
+  bool FinalizeResult(void** tag, bool* status) {
+    *status = true;
+    return true;
+  }
+};
+
 class Server::SyncRequest final : public CompletionQueueTag {
  public:
   SyncRequest(RpcServiceMethod* method, void* tag)
@@ -145,7 +153,7 @@ class Server::SyncRequest final : public CompletionQueueTag {
     grpc_metadata_array_destroy(&request_metadata_);
   }
 
-  void SetupRequest() { cq_ = grpc_completion_queue_create(nullptr); }
+  void SetupRequest() { cq_ = grpc_completion_queue_create_for_pluck(nullptr); }
 
   void TeardownRequest() {
     grpc_completion_queue_destroy(cq_);
@@ -213,10 +221,15 @@ class Server::SyncRequest final : public CompletionQueueTag {
           MethodHandler::HandlerParameter(&call_, &ctx_, request_payload_));
       global_callbacks->PostSynchronousRequest(&ctx_);
       request_payload_ = nullptr;
-      void* ignored_tag;
-      bool ignored_ok;
+
       cq_.Shutdown();
-      GPR_ASSERT(cq_.Next(&ignored_tag, &ignored_ok) == false);
+
+      CompletionQueueTag* op_tag = ctx_.GetCompletionOpTag();
+      cq_.TryPluck(op_tag, gpr_inf_future(GPR_CLOCK_REALTIME));
+
+      /* Ensure the cq_ is shutdown */
+      DummyTag ignored_tag;
+      GPR_ASSERT(cq_.Pluck(&ignored_tag) == false);
     }
 
    private:

src/cpp/server/server_context.cc

@@ -167,6 +167,10 @@ void ServerContext::BeginCompletionOp(Call* call) {
   call->PerformOps(completion_op_);
 }
 
+CompletionQueueTag* ServerContext::GetCompletionOpTag() {
+  return static_cast<CompletionQueueTag*>(completion_op_);
+}
+
 void ServerContext::AddInitialMetadata(const grpc::string& key,
                                        const grpc::string& value) {
   initial_metadata_.insert(std::make_pair(key, value));

src/csharp/Grpc.Core.Tests/Internal/CompletionQueueSafeHandleTest.cs

@@ -43,19 +43,19 @@ namespace Grpc.Core.Internal.Tests
     public class CompletionQueueSafeHandleTest
     {
         [Test]
-        public void CreateAndDestroy()
+        public void CreateSyncAndDestroy()
         {
             GrpcEnvironment.AddRef();
-            var cq = CompletionQueueSafeHandle.Create();
+            var cq = CompletionQueueSafeHandle.CreateSync();
             cq.Dispose();
             GrpcEnvironment.ReleaseAsync().Wait();
         }
 
         [Test]
-        public void CreateAndShutdown()
+        public void CreateAsyncAndShutdown()
         {
-            GrpcEnvironment.AddRef();
-            var cq = CompletionQueueSafeHandle.Create();
+            var env = GrpcEnvironment.AddRef();
+            var cq = CompletionQueueSafeHandle.CreateAsync(new CompletionRegistry(env));
             cq.Shutdown();
             var ev = cq.Next();
             cq.Dispose();

src/csharp/Grpc.Core.Tests/PInvokeTest.cs

@@ -53,7 +53,7 @@ namespace Grpc.Core.Tests
         /// (~1.26us .NET Windows)
         /// </summary>
         [Test]
-        public void CompletionQueueCreateDestroyBenchmark()
+        public void CompletionQueueCreateSyncDestroyBenchmark()
         {
             GrpcEnvironment.AddRef();  // completion queue requires gRPC environment being initialized.
 
@@ -61,7 +61,7 @@ namespace Grpc.Core.Tests
                 10, 10,
                 () =>
                 {
-                    CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.Create();
+                    CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.CreateSync();
                     cq.Dispose();
                 });
 

src/csharp/Grpc.Core/Internal/AsyncCall.cs

@@ -87,7 +87,7 @@ namespace Grpc.Core.Internal
             var profiler = Profilers.ForCurrentThread();
 
             using (profiler.NewScope("AsyncCall.UnaryCall"))
-            using (CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.Create())
+            using (CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.CreateSync())
             {
                 byte[] payload = UnsafeSerialize(msg);
 

src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs

@@ -51,14 +51,20 @@ namespace Grpc.Core.Internal
         {
         }
 
-        public static CompletionQueueSafeHandle Create()
+        /// <summary>
+        /// Create a completion queue that can only be used for Pluck operations.
+        /// </summary>
+        public static CompletionQueueSafeHandle CreateSync()
         {
-            return Native.grpcsharp_completion_queue_create();
+            return Native.grpcsharp_completion_queue_create_sync();
         }
 
-        public static CompletionQueueSafeHandle Create(CompletionRegistry completionRegistry)
+        /// <summary>
+        /// Create a completion queue that can only be used for Next operations.
+        /// </summary>
+        public static CompletionQueueSafeHandle CreateAsync(CompletionRegistry completionRegistry)
         {
-            var cq = Native.grpcsharp_completion_queue_create();
+            var cq = Native.grpcsharp_completion_queue_create_async();
             cq.completionRegistry = completionRegistry;
             return cq;
         }

src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs

@@ -197,7 +197,7 @@ namespace Grpc.Core.Internal
             for (int i = 0; i < completionQueueCount; i++)
             {
                 var completionRegistry = new CompletionRegistry(environment);
-                list.Add(CompletionQueueSafeHandle.Create(completionRegistry));
+                list.Add(CompletionQueueSafeHandle.CreateAsync(completionRegistry));
             }
             return list.AsReadOnly();
         }

src/csharp/Grpc.Core/Internal/NativeMethods.cs

@@ -115,7 +115,8 @@ namespace Grpc.Core.Internal
 
         public readonly Delegates.grpcsharp_sizeof_grpc_event_delegate grpcsharp_sizeof_grpc_event;
 
-        public readonly Delegates.grpcsharp_completion_queue_create_delegate grpcsharp_completion_queue_create;
+        public readonly Delegates.grpcsharp_completion_queue_create_async_delegate grpcsharp_completion_queue_create_async;
+        public readonly Delegates.grpcsharp_completion_queue_create_sync_delegate grpcsharp_completion_queue_create_sync;
         public readonly Delegates.grpcsharp_completion_queue_shutdown_delegate grpcsharp_completion_queue_shutdown;
         public readonly Delegates.grpcsharp_completion_queue_next_delegate grpcsharp_completion_queue_next;
         public readonly Delegates.grpcsharp_completion_queue_pluck_delegate grpcsharp_completion_queue_pluck;
@@ -229,7 +230,8 @@ namespace Grpc.Core.Internal
 
             this.grpcsharp_sizeof_grpc_event = GetMethodDelegate<Delegates.grpcsharp_sizeof_grpc_event_delegate>(library);
 
-            this.grpcsharp_completion_queue_create = GetMethodDelegate<Delegates.grpcsharp_completion_queue_create_delegate>(library);
+            this.grpcsharp_completion_queue_create_async = GetMethodDelegate<Delegates.grpcsharp_completion_queue_create_async_delegate>(library);
+            this.grpcsharp_completion_queue_create_sync = GetMethodDelegate<Delegates.grpcsharp_completion_queue_create_sync_delegate>(library);
             this.grpcsharp_completion_queue_shutdown = GetMethodDelegate<Delegates.grpcsharp_completion_queue_shutdown_delegate>(library);
             this.grpcsharp_completion_queue_next = GetMethodDelegate<Delegates.grpcsharp_completion_queue_next_delegate>(library);
             this.grpcsharp_completion_queue_pluck = GetMethodDelegate<Delegates.grpcsharp_completion_queue_pluck_delegate>(library);
@@ -383,7 +385,8 @@ namespace Grpc.Core.Internal
 
             public delegate int grpcsharp_sizeof_grpc_event_delegate();
 
-            public delegate CompletionQueueSafeHandle grpcsharp_completion_queue_create_delegate();
+            public delegate CompletionQueueSafeHandle grpcsharp_completion_queue_create_async_delegate();
+            public delegate CompletionQueueSafeHandle grpcsharp_completion_queue_create_sync_delegate();
             public delegate void grpcsharp_completion_queue_shutdown_delegate(CompletionQueueSafeHandle cq);
             public delegate CompletionQueueEvent grpcsharp_completion_queue_next_delegate(CompletionQueueSafeHandle cq);
             public delegate CompletionQueueEvent grpcsharp_completion_queue_pluck_delegate(CompletionQueueSafeHandle cq, IntPtr tag);

src/csharp/ext/grpc_csharp_ext.c

@@ -354,8 +354,13 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_shutdown(void) { grpc_shutdown(); }
 /* Completion queue */
 
 GPR_EXPORT grpc_completion_queue *GPR_CALLTYPE
-grpcsharp_completion_queue_create(void) {
-  return grpc_completion_queue_create(NULL);
+grpcsharp_completion_queue_create_async(void) {
+  return grpc_completion_queue_create_for_next(NULL);
+}
+
+GPR_EXPORT grpc_completion_queue *GPR_CALLTYPE
+grpcsharp_completion_queue_create_sync(void) {
+  return grpc_completion_queue_create_for_pluck(NULL);
 }
 
 GPR_EXPORT void GPR_CALLTYPE

src/node/ext/completion_queue_threadpool.cc

@@ -34,14 +34,14 @@
 /* I don't like using #ifndef, but I don't see a better way to do this */
 #ifndef GRPC_UV
 
-#include <node.h>
 #include <nan.h>
+#include <node.h>
 
+#include "call.h"
+#include "completion_queue.h"
 #include "grpc/grpc.h"
 #include "grpc/support/log.h"
 #include "grpc/support/time.h"
-#include "completion_queue.h"
-#include "call.h"
 
 namespace grpc {
 namespace node {
@@ -111,8 +111,8 @@ CompletionQueueAsyncWorker::CompletionQueueAsyncWorker()
 CompletionQueueAsyncWorker::~CompletionQueueAsyncWorker() {}
 
 void CompletionQueueAsyncWorker::Execute() {
-  result =
-      grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+  result = grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME),
+                                      NULL);
   if (!result.success) {
     SetErrorMessage("The async function encountered an error");
   }
@@ -141,7 +141,7 @@ void CompletionQueueAsyncWorker::Init(Local<Object> exports) {
   Nan::HandleScope scope;
   current_threads = 0;
   waiting_next_calls = 0;
-  queue = grpc_completion_queue_create(NULL);
+  queue = grpc_completion_queue_create_for_next(NULL);
 }
 
 void CompletionQueueAsyncWorker::HandleOKCallback() {
@@ -168,9 +168,7 @@ grpc_completion_queue *GetCompletionQueue() {
   return CompletionQueueAsyncWorker::GetQueue();
 }
 
-void CompletionQueueNext() {
-  CompletionQueueAsyncWorker::Next();
-}
+void CompletionQueueNext() { CompletionQueueAsyncWorker::Next(); }
 
 void CompletionQueueInit(Local<Object> exports) {
   CompletionQueueAsyncWorker::Init(exports);
@@ -179,4 +177,4 @@ void CompletionQueueInit(Local<Object> exports) {
 }  // namespace node
 }  // namespace grpc
 
-#endif  /* GRPC_UV */
+#endif /* GRPC_UV */

src/node/ext/completion_queue_uv.cc

@@ -33,10 +33,10 @@
 
 #ifdef GRPC_UV
 
-#include <uv.h>
+#include <grpc/grpc.h>
 #include <node.h>
+#include <uv.h>
 #include <v8.h>
-#include <grpc/grpc.h>
 
 #include "call.h"
 #include "completion_queue.h"
@@ -57,8 +57,8 @@ void drain_completion_queue(uv_prepare_t *handle) {
   grpc_event event;
   (void)handle;
   do {
-    event = grpc_completion_queue_next(
-        queue, gpr_inf_past(GPR_CLOCK_MONOTONIC), NULL);
+    event = grpc_completion_queue_next(queue, gpr_inf_past(GPR_CLOCK_MONOTONIC),
+                                       NULL);
 
     if (event.type == GRPC_OP_COMPLETE) {
       const char *error_message;
@@ -77,9 +77,7 @@ void drain_completion_queue(uv_prepare_t *handle) {
   } while (event.type != GRPC_QUEUE_TIMEOUT);
 }
 
-grpc_completion_queue *GetCompletionQueue() {
-  return queue;
-}
+grpc_completion_queue *GetCompletionQueue() { return queue; }
 
 void CompletionQueueNext() {
   if (pending_batches == 0) {
@@ -90,7 +88,7 @@ void CompletionQueueNext() {
 }
 
 void CompletionQueueInit(Local<Object> exports) {
-  queue = grpc_completion_queue_create(NULL);
+  queue = grpc_completion_queue_create_for_next(NULL);
   uv_prepare_init(uv_default_loop(), &prepare);
   pending_batches = 0;
 }

src/node/ext/server_generic.cc

@@ -35,8 +35,8 @@
 
 #include "server.h"
 
-#include <node.h>
 #include <nan.h>
+#include <node.h>
 #include "grpc/grpc.h"
 #include "grpc/support/time.h"
 
@@ -44,7 +44,7 @@ namespace grpc {
 namespace node {
 
 Server::Server(grpc_server *server) : wrapped_server(server) {
-  shutdown_queue = grpc_completion_queue_create(NULL);
+  shutdown_queue = grpc_completion_queue_create_for_pluck(NULL);
   grpc_server_register_non_listening_completion_queue(server, shutdown_queue,
                                                       NULL);
 }

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

@@ -48,7 +48,7 @@
 
 - (instancetype)init {
   if ((self = [super init])) {
-    _unmanagedQueue = grpc_completion_queue_create(NULL);
+    _unmanagedQueue = grpc_completion_queue_create_for_next(NULL);
 
     // This is for the following block to capture the pointer by value (instead
     // of retaining self and doing self->_unmanagedQueue). This is essential

src/objective-c/tests/CoreCronetEnd2EndTests/CoreCronetEnd2EndTests.m

@@ -79,7 +79,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
   gpr_join_host_port(&ffd->localaddr, "127.0.0.1", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

src/objective-c/tests/CronetUnitTests/CronetUnitTests.m

@@ -160,7 +160,7 @@ unsigned int parse_h2_length(const char *field) {
   int port = grpc_pick_unused_port_or_die();
   char *addr;
   gpr_join_host_port(&addr, "127.0.0.1", port);
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
   stream_engine *cronetEngine = [Cronet getGlobalEngine];
   grpc_channel *client =
       grpc_cronet_secure_channel_create(cronetEngine, addr, NULL, NULL);
@@ -295,7 +295,7 @@ unsigned int parse_h2_length(const char *field) {
   int port = grpc_pick_unused_port_or_die();
   char *addr;
   gpr_join_host_port(&addr, "127.0.0.1", port);
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
   stream_engine *cronetEngine = [Cronet getGlobalEngine];
   grpc_channel *client =
       grpc_cronet_secure_channel_create(cronetEngine, addr, args, NULL);

src/php/ext/grpc/completion_queue.c

@@ -38,13 +38,10 @@
 grpc_completion_queue *completion_queue;
 
 void grpc_php_init_completion_queue(TSRMLS_D) {
-  completion_queue = grpc_completion_queue_create(NULL);
+  completion_queue = grpc_completion_queue_create_for_pluck(NULL);
 }
 
 void grpc_php_shutdown_completion_queue(TSRMLS_D) {
   grpc_completion_queue_shutdown(completion_queue);
-  while (grpc_completion_queue_next(completion_queue,
-                                    gpr_inf_future(GPR_CLOCK_REALTIME),
-                                    NULL).type != GRPC_QUEUE_SHUTDOWN);
   grpc_completion_queue_destroy(completion_queue);
 }

src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi

@@ -40,7 +40,7 @@ cdef class CompletionQueue:
   def __cinit__(self):
     grpc_init()
     with nogil:
-      self.c_completion_queue = grpc_completion_queue_create(NULL)
+      self.c_completion_queue = grpc_completion_queue_create_for_next(NULL)
     self.is_shutting_down = False
     self.is_shutdown = False
 

src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi

@@ -309,7 +309,8 @@ cdef extern from "grpc/grpc.h":
   void grpc_init() nogil
   void grpc_shutdown() nogil
 
-  grpc_completion_queue *grpc_completion_queue_create(void *reserved) nogil
+  grpc_completion_queue *grpc_completion_queue_create_for_next(void *reserved) nogil
+
   grpc_event grpc_completion_queue_next(grpc_completion_queue *cq,
                                         gpr_timespec deadline,
                                         void *reserved) nogil

src/ruby/ext/grpc/rb_channel.c

@@ -351,7 +351,7 @@ static VALUE grpc_rb_channel_create_call(VALUE self, VALUE parent, VALUE mask,
     parent_call = grpc_rb_get_wrapped_call(parent);
   }
 
-  cq = grpc_completion_queue_create(NULL);
+  cq = grpc_completion_queue_create_for_pluck(NULL);
   TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper);
   ch = wrapper->wrapped;
   if (ch == NULL) {
@@ -522,7 +522,7 @@ static VALUE run_poll_channels_loop(VALUE arg) {
  * TODO(apolcyn) remove this when core handles new RPCs on dead connections.
  */
 static void start_poll_channels_loop() {
-  channel_polling_cq = grpc_completion_queue_create(NULL);
+  channel_polling_cq = grpc_completion_queue_create_for_next(NULL);
   gpr_mu_init(&global_connection_polling_mu);
   abort_channel_polling = 0;
   rb_thread_create(run_poll_channels_loop, NULL);

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

@@ -233,7 +233,7 @@ 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_type)(void *reserved);
+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
 typedef grpc_event(*grpc_completion_queue_next_type)(grpc_completion_queue *cq, gpr_timespec deadline, void *reserved);

src/ruby/ext/grpc/rb_server.c

@@ -37,15 +37,15 @@
 #include "rb_server.h"
 
 #include <grpc/grpc.h>
-#include <grpc/support/atm.h>
 #include <grpc/grpc_security.h>
+#include <grpc/support/atm.h>
 #include <grpc/support/log.h>
+#include "rb_byte_buffer.h"
 #include "rb_call.h"
 #include "rb_channel_args.h"
 #include "rb_completion_queue.h"
-#include "rb_server_credentials.h"
-#include "rb_byte_buffer.h"
 #include "rb_grpc.h"
+#include "rb_server_credentials.h"
 
 /* grpc_rb_cServer is the ruby class that proxies grpc_server. */
 static VALUE grpc_rb_cServer = Qnil;
@@ -93,9 +93,8 @@ static void grpc_rb_server_free(void *p) {
   };
   svr = (grpc_rb_server *)p;
 
-  deadline = gpr_time_add(
-      gpr_now(GPR_CLOCK_REALTIME),
-      gpr_time_from_seconds(2, GPR_TIMESPAN));
+  deadline = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+                          gpr_time_from_seconds(2, GPR_TIMESPAN));
 
   destroy_server(svr, deadline);
 
@@ -104,13 +103,15 @@ static void grpc_rb_server_free(void *p) {
 
 static const rb_data_type_t grpc_rb_server_data_type = {
     "grpc_server",
-    {GRPC_RB_GC_NOT_MARKED, grpc_rb_server_free, GRPC_RB_MEMSIZE_UNAVAILABLE,
+    {GRPC_RB_GC_NOT_MARKED,
+     grpc_rb_server_free,
+     GRPC_RB_MEMSIZE_UNAVAILABLE,
      {NULL, NULL}},
     NULL,
     NULL,
 #ifdef RUBY_TYPED_FREE_IMMEDIATELY
-    /* It is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because the free function would block
-     * and we might want to unlock GVL
+    /* It is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because the free
+     * function would block and we might want to unlock GVL
      * TODO(yugui) Unlock GVL?
      */
     0,
@@ -131,7 +132,7 @@ static VALUE grpc_rb_server_alloc(VALUE cls) {
 
   Initializes server instances. */
 static VALUE grpc_rb_server_init(VALUE self, VALUE channel_args) {
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_pluck(NULL);
   grpc_rb_server *wrapper = NULL;
   grpc_server *srv = NULL;
   grpc_channel_args args;
@@ -163,7 +164,7 @@ typedef struct request_call_stack {
 
 /* grpc_request_call_stack_init ensures the request_call_stack is properly
  * initialized */
-static void grpc_request_call_stack_init(request_call_stack* st) {
+static void grpc_request_call_stack_init(request_call_stack *st) {
   MEMZERO(st, request_call_stack, 1);
   grpc_metadata_array_init(&st->md_ary);
   grpc_call_details_init(&st->details);
@@ -171,7 +172,7 @@ static void grpc_request_call_stack_init(request_call_stack* st) {
 
 /* grpc_request_call_stack_cleanup ensures the request_call_stack is properly
  * cleaned up */
-static void grpc_request_call_stack_cleanup(request_call_stack* st) {
+static void grpc_request_call_stack_cleanup(request_call_stack *st) {
   grpc_metadata_array_destroy(&st->md_ary);
   grpc_call_details_destroy(&st->details);
 }
@@ -187,8 +188,9 @@ static VALUE grpc_rb_server_request_call(VALUE self) {
   grpc_call_error err;
   request_call_stack st;
   VALUE result;
-  void *tag = (void*)&st;
-  grpc_completion_queue *call_queue = grpc_completion_queue_create(NULL);
+  void *tag = (void *)&st;
+  grpc_completion_queue *call_queue =
+      grpc_completion_queue_create_for_pluck(NULL);
   gpr_timespec deadline;
 
   TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
@@ -199,9 +201,8 @@ static VALUE grpc_rb_server_request_call(VALUE self) {
   grpc_request_call_stack_init(&st);
   /* call grpc_server_request_call, then wait for it to complete using
    * pluck_event */
-  err = grpc_server_request_call(
-      s->wrapped, &call, &st.details, &st.md_ary,
-      call_queue, s->queue, tag);
+  err = grpc_server_request_call(s->wrapped, &call, &st.details, &st.md_ary,
+                                 call_queue, s->queue, tag);
   if (err != GRPC_CALL_OK) {
     grpc_request_call_stack_cleanup(&st);
     rb_raise(grpc_rb_eCallError,
@@ -218,8 +219,6 @@ static VALUE grpc_rb_server_request_call(VALUE self) {
     return Qnil;
   }
 
-
-
   /* build the NewServerRpc struct result */
   deadline = gpr_convert_clock_type(st.details.deadline, GPR_CLOCK_REALTIME);
   result = rb_struct_new(
@@ -299,8 +298,7 @@ static VALUE grpc_rb_server_add_http2_port(VALUE self, VALUE port,
     return Qnil;
   } else if (TYPE(rb_creds) == T_SYMBOL) {
     if (id_insecure_server != SYM2ID(rb_creds)) {
-      rb_raise(rb_eTypeError,
-               "bad creds symbol, want :this_port_is_insecure");
+      rb_raise(rb_eTypeError, "bad creds symbol, want :this_port_is_insecure");
       return Qnil;
     }
     recvd_port =
@@ -312,9 +310,8 @@ static VALUE grpc_rb_server_add_http2_port(VALUE self, VALUE port,
     }
   } else {
     creds = grpc_rb_get_wrapped_server_credentials(rb_creds);
-    recvd_port =
-        grpc_server_add_secure_http2_port(s->wrapped, StringValueCStr(port),
-                                          creds);
+    recvd_port = grpc_server_add_secure_http2_port(
+        s->wrapped, StringValueCStr(port), creds);
     if (recvd_port == 0) {
       rb_raise(rb_eRuntimeError,
                "could not add secure port %s to server, not sure why",
@@ -333,18 +330,17 @@ void Init_grpc_server() {
 
   /* Provides a ruby constructor and support for dup/clone. */
   rb_define_method(grpc_rb_cServer, "initialize", grpc_rb_server_init, 1);
-  rb_define_method(grpc_rb_cServer, "initialize_copy",
-                   grpc_rb_cannot_init_copy, 1);
+  rb_define_method(grpc_rb_cServer, "initialize_copy", grpc_rb_cannot_init_copy,
+                   1);
 
   /* Add the server methods. */
-  rb_define_method(grpc_rb_cServer, "request_call",
-                   grpc_rb_server_request_call, 0);
+  rb_define_method(grpc_rb_cServer, "request_call", grpc_rb_server_request_call,
+                   0);
   rb_define_method(grpc_rb_cServer, "start", grpc_rb_server_start, 0);
   rb_define_method(grpc_rb_cServer, "destroy", grpc_rb_server_destroy, -1);
   rb_define_alias(grpc_rb_cServer, "close", "destroy");
   rb_define_method(grpc_rb_cServer, "add_http2_port",
-                   grpc_rb_server_add_http2_port,
-                   2);
+                   grpc_rb_server_add_http2_port, 2);
   id_at = rb_intern("at");
   id_insecure_server = rb_intern("this_port_is_insecure");
 }

test/core/bad_client/bad_client.c

@@ -104,6 +104,7 @@ void grpc_run_bad_client_test(
   grpc_slice_buffer outgoing;
   grpc_closure done_write_closure;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_completion_queue *shutdown_cq;
 
   hex = gpr_dump(client_payload, client_payload_length,
                  GPR_DUMP_HEX | GPR_DUMP_ASCII);
@@ -121,7 +122,7 @@ void grpc_run_bad_client_test(
 
   /* Create server, completion events */
   a.server = grpc_server_create(NULL, NULL);
-  a.cq = grpc_completion_queue_create(NULL);
+  a.cq = grpc_completion_queue_create_for_next(NULL);
   gpr_event_init(&a.done_thd);
   gpr_event_init(&a.done_write);
   a.validator = server_validator;
@@ -194,10 +195,13 @@ void grpc_run_bad_client_test(
     grpc_endpoint_destroy(&exec_ctx, sfd.client);
     grpc_exec_ctx_finish(&exec_ctx);
   }
-  grpc_server_shutdown_and_notify(a.server, a.cq, NULL);
+
+  shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
+  grpc_server_shutdown_and_notify(a.server, shutdown_cq, NULL);
   GPR_ASSERT(grpc_completion_queue_pluck(
-                 a.cq, NULL, grpc_timeout_seconds_to_deadline(1), NULL)
+                 shutdown_cq, NULL, grpc_timeout_seconds_to_deadline(1), NULL)
                  .type == GRPC_OP_COMPLETE);
+  grpc_completion_queue_destroy(shutdown_cq);
   grpc_server_destroy(a.server);
   grpc_completion_queue_destroy(a.cq);
   grpc_slice_buffer_destroy_internal(&exec_ctx, &outgoing);

test/core/bad_ssl/bad_ssl_test.c

@@ -61,7 +61,7 @@ static void run_test(const char *target, size_t nops) {
   grpc_status_code status;
   grpc_call_error error;
   gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
   cq_verifier *cqv = cq_verifier_create(cq);
 
   grpc_op ops[6];

test/core/bad_ssl/server_common.c

@@ -66,7 +66,9 @@ void bad_ssl_run(grpc_server *server) {
   grpc_call *s = NULL;
   grpc_call_details call_details;
   grpc_metadata_array request_metadata_recv;
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
+  grpc_completion_queue *shutdown_cq;
 
   grpc_call_details_init(&call_details);
   grpc_metadata_array_init(&request_metadata_recv);
@@ -82,10 +84,13 @@ void bad_ssl_run(grpc_server *server) {
   while (!shutdown_finished) {
     if (got_sigint && !shutdown_started) {
       gpr_log(GPR_INFO, "Shutting down due to SIGINT");
-      grpc_server_shutdown_and_notify(server, cq, NULL);
-      GPR_ASSERT(grpc_completion_queue_pluck(
-                     cq, NULL, grpc_timeout_seconds_to_deadline(5), NULL)
-                     .type == GRPC_OP_COMPLETE);
+      shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
+      grpc_server_shutdown_and_notify(server, shutdown_cq, NULL);
+      GPR_ASSERT(
+          grpc_completion_queue_pluck(shutdown_cq, NULL,
+                                      grpc_timeout_seconds_to_deadline(5), NULL)
+              .type == GRPC_OP_COMPLETE);
+      grpc_completion_queue_destroy(shutdown_cq);
       grpc_completion_queue_shutdown(cq);
       shutdown_started = 1;
     }

test/core/client_channel/lb_policies_test.c

@@ -59,6 +59,7 @@ typedef struct servers_fixture {
   grpc_server **servers;
   grpc_call **server_calls;
   grpc_completion_queue *cq;
+  grpc_completion_queue *shutdown_cq;
   char **servers_hostports;
   grpc_metadata_array *request_metadata_recv;
 } servers_fixture;
@@ -146,10 +147,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 static void kill_server(const servers_fixture *f, size_t i) {
   gpr_log(GPR_INFO, "KILLING SERVER %" PRIuPTR, i);
   GPR_ASSERT(f->servers[i] != NULL);
-  grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000));
-  GPR_ASSERT(
-      grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000), NULL)
-          .type == GRPC_OP_COMPLETE);
+  grpc_server_shutdown_and_notify(f->servers[i], f->shutdown_cq, tag(10000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(10000),
+                                         n_millis_time(5000), NULL)
+                 .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->servers[i]);
   f->servers[i] = NULL;
 }
@@ -196,7 +197,8 @@ static servers_fixture *setup_servers(const char *server_host,
   /* Create servers. */
   f->servers = gpr_malloc(sizeof(grpc_server *) * num_servers);
   f->servers_hostports = gpr_malloc(sizeof(char *) * num_servers);
-  f->cq = grpc_completion_queue_create(NULL);
+  f->cq = grpc_completion_queue_create_for_next(NULL);
+  f->shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
   for (i = 0; i < num_servers; i++) {
     grpc_metadata_array_init(&f->request_metadata_recv[i]);
     gpr_join_host_port(&f->servers_hostports[i], server_host,
@@ -212,8 +214,8 @@ static void teardown_servers(servers_fixture *f) {
   /* Destroy server. */
   for (i = 0; i < f->num_servers; i++) {
     if (f->servers[i] == NULL) continue;
-    grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000));
-    GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000),
+    grpc_server_shutdown_and_notify(f->servers[i], f->shutdown_cq, tag(10000));
+    GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(10000),
                                            n_millis_time(5000), NULL)
                    .type == GRPC_OP_COMPLETE);
     grpc_server_destroy(f->servers[i]);
@@ -221,6 +223,7 @@ static void teardown_servers(servers_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 
   gpr_free(f->servers);
 

test/core/end2end/bad_server_response_test.c

@@ -178,7 +178,7 @@ static void start_rpc(int target_port, grpc_status_code expected_status,
   cq_verifier *cqv;
   grpc_slice details;
 
-  state.cq = grpc_completion_queue_create(NULL);
+  state.cq = grpc_completion_queue_create_for_next(NULL);
   cqv = cq_verifier_create(state.cq);
   gpr_join_host_port(&state.target, "127.0.0.1", target_port);
   state.channel = grpc_insecure_channel_create(state.target, NULL, NULL);

test/core/end2end/connection_refused_test.c

@@ -68,7 +68,7 @@ static void run_test(bool wait_for_ready, bool use_service_config) {
 
   grpc_metadata_array_init(&trailing_metadata_recv);
 
-  cq = grpc_completion_queue_create(NULL);
+  cq = grpc_completion_queue_create_for_next(NULL);
   cqv = cq_verifier_create(cq);
 
   /* if using service config, create channel args */

test/core/end2end/dualstack_socket_test.c

@@ -76,6 +76,7 @@ void test_connect(const char *server_host, const char *client_host, int port,
   grpc_channel *client;
   grpc_server *server;
   grpc_completion_queue *cq;
+  grpc_completion_queue *shutdown_cq;
   grpc_call *c;
   grpc_call *s;
   cq_verifier *cqv;
@@ -107,7 +108,7 @@ void test_connect(const char *server_host, const char *client_host, int port,
   grpc_call_details_init(&call_details);
 
   /* Create server. */
-  cq = grpc_completion_queue_create(NULL);
+  cq = grpc_completion_queue_create_for_next(NULL);
   server = grpc_server_create(NULL, NULL);
   grpc_server_register_completion_queue(server, cq, NULL);
   GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port(
@@ -259,11 +260,14 @@ void test_connect(const char *server_host, const char *client_host, int port,
   grpc_channel_destroy(client);
 
   /* Destroy server. */
-  grpc_server_shutdown_and_notify(server, cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
+  grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(server);
+  grpc_completion_queue_destroy(shutdown_cq);
   grpc_completion_queue_shutdown(cq);
   drain_cq(cq);
   grpc_completion_queue_destroy(cq);

test/core/end2end/end2end_tests.h

@@ -53,6 +53,7 @@ typedef struct grpc_end2end_test_config grpc_end2end_test_config;
 
 struct grpc_end2end_test_fixture {
   grpc_completion_queue *cq;
+  grpc_completion_queue *shutdown_cq;
   grpc_server *server;
   grpc_channel *client;
   void *fixture_data;

test/core/end2end/fixtures/h2_census.c

@@ -65,7 +65,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_compress.c

@@ -69,7 +69,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack_compression(
 
   memset(&f, 0, sizeof(f));
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_fakesec.c

@@ -60,7 +60,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_fd.c

@@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   grpc_end2end_test_fixture f;
   memset(&f, 0, sizeof(f));
   f.fixture_data = fixture_data;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   create_sockets(fixture_data->fd_pair);
 

test/core/end2end/fixtures/h2_full+pipe.c

@@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_full+trace.c

@@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_full.c

@@ -64,7 +64,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_http_proxy.c

@@ -69,7 +69,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
   ffd->proxy = grpc_end2end_http_proxy_create();
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_load_reporting.c

@@ -67,7 +67,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_load_reporting(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_oauth2.c

@@ -113,7 +113,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_proxy.c

@@ -79,7 +79,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
   ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_sockpair+trace.c

@@ -94,7 +94,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   grpc_end2end_test_fixture f;
   memset(&f, 0, sizeof(f));
   f.fixture_data = sfd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   *sfd = grpc_iomgr_create_endpoint_pair("fixture", NULL);
 

test/core/end2end/fixtures/h2_sockpair.c

@@ -88,7 +88,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   grpc_end2end_test_fixture f;
   memset(&f, 0, sizeof(f));
   f.fixture_data = sfd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   *sfd = grpc_iomgr_create_endpoint_pair("fixture", NULL);
 

test/core/end2end/fixtures/h2_sockpair_1byte.c

@@ -88,7 +88,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   grpc_end2end_test_fixture f;
   memset(&f, 0, sizeof(f));
   f.fixture_data = sfd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   grpc_arg a[] = {{.key = GRPC_ARG_TCP_READ_CHUNK_SIZE,
                    .type = GRPC_ARG_INTEGER,

test/core/end2end/fixtures/h2_ssl.c

@@ -64,7 +64,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_ssl_cert.c

@@ -67,7 +67,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
   gpr_join_host_port(&ffd->localaddr, "localhost", port);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }
@@ -289,9 +290,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -310,6 +312,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_fixture f,

test/core/end2end/fixtures/h2_ssl_proxy.c

@@ -100,7 +100,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
   ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/h2_uds.c

@@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
                unique++);
 
   f.fixture_data = ffd;
-  f.cq = grpc_completion_queue_create(NULL);
+  f.cq = grpc_completion_queue_create_for_next(NULL);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
 
   return f;
 }

test/core/end2end/fixtures/proxy.c

@@ -104,7 +104,7 @@ grpc_end2end_proxy *grpc_end2end_proxy_create(const grpc_end2end_proxy_def *def,
   gpr_log(GPR_DEBUG, "PROXY ADDR:%s BACKEND:%s", proxy->proxy_port,
           proxy->server_port);
 
-  proxy->cq = grpc_completion_queue_create(NULL);
+  proxy->cq = grpc_completion_queue_create_for_next(NULL);
   proxy->server = def->create_server(proxy->proxy_port, server_args);
   proxy->client = def->create_client(proxy->server_port, client_args);
 

test/core/end2end/fuzzers/api_fuzzer.c

@@ -735,7 +735,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
   g_active_call = new_call(NULL, ROOT);
   g_resource_quota = grpc_resource_quota_create("api_fuzzer");
 
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
 
   while (!is_eof(&inp) || g_channel != NULL || g_server != NULL ||
          pending_channel_watches > 0 || pending_pings > 0 ||

test/core/end2end/fuzzers/client_fuzzer.c

@@ -65,7 +65,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
       grpc_mock_endpoint_create(discard_write, resource_quota);
   grpc_resource_quota_unref_internal(&exec_ctx, resource_quota);
 
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
   grpc_transport *transport =
       grpc_create_chttp2_transport(&exec_ctx, NULL, mock_endpoint, 1);
   grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);

test/core/end2end/fuzzers/server_fuzzer.c

@@ -67,7 +67,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
       grpc_slice_from_copied_buffer((const char *)data, size));
 
   grpc_server *server = grpc_server_create(NULL, NULL);
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
   grpc_server_register_completion_queue(server, cq, NULL);
   // TODO(ctiller): add registered methods (one for POST, one for PUT)
   // void *registered_method =

test/core/end2end/goaway_server_test.c

@@ -129,7 +129,7 @@ int main(int argc, char **argv) {
   grpc_metadata_array_init(&request_metadata2);
   grpc_call_details_init(&request_details2);
 
-  cq = grpc_completion_queue_create(NULL);
+  cq = grpc_completion_queue_create_for_next(NULL);
   cqv = cq_verifier_create(cq);
 
   /* reserve two ports */

test/core/end2end/invalid_call_argument_test.c

@@ -75,7 +75,7 @@ static void prepare_test(int is_client) {
   grpc_metadata_array_init(&g_state.initial_metadata_recv);
   grpc_metadata_array_init(&g_state.trailing_metadata_recv);
   g_state.deadline = grpc_timeout_seconds_to_deadline(2);
-  g_state.cq = grpc_completion_queue_create(NULL);
+  g_state.cq = grpc_completion_queue_create_for_next(NULL);
   g_state.cqv = cq_verifier_create(g_state.cq);
   g_state.details = grpc_empty_slice();
   memset(g_state.ops, 0, sizeof(g_state.ops));
@@ -125,6 +125,7 @@ static void prepare_test(int is_client) {
 }
 
 static void cleanup_test() {
+  grpc_completion_queue *shutdown_cq;
   grpc_call_destroy(g_state.call);
   cq_verifier_destroy(g_state.cqv);
   grpc_channel_destroy(g_state.chan);
@@ -133,12 +134,14 @@ static void cleanup_test() {
   grpc_metadata_array_destroy(&g_state.trailing_metadata_recv);
 
   if (!g_state.is_client) {
+    shutdown_cq = grpc_completion_queue_create_for_pluck(NULL);
     grpc_call_destroy(g_state.server_call);
-    grpc_server_shutdown_and_notify(g_state.server, g_state.cq, tag(1000));
-    GPR_ASSERT(grpc_completion_queue_pluck(g_state.cq, tag(1000),
+    grpc_server_shutdown_and_notify(g_state.server, shutdown_cq, tag(1000));
+    GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000),
                                            grpc_timeout_seconds_to_deadline(5),
                                            NULL)
                    .type == GRPC_OP_COMPLETE);
+    grpc_completion_queue_destroy(shutdown_cq);
     grpc_server_destroy(g_state.server);
     grpc_call_details_destroy(&g_state.call_details);
     grpc_metadata_array_destroy(&g_state.server_initial_metadata_recv);

test/core/end2end/multiple_server_queues_test.c

@@ -37,27 +37,50 @@
 int main(int argc, char **argv) {
   grpc_completion_queue *cq1;
   grpc_completion_queue *cq2;
+  grpc_completion_queue *cq3;
+  grpc_completion_queue_attributes attr;
+
   grpc_server *server;
 
   grpc_test_init(argc, argv);
   grpc_init();
-  cq1 = grpc_completion_queue_create(NULL);
-  cq2 = grpc_completion_queue_create(NULL);
+
+  attr.version = 1;
+  attr.cq_completion_type = GRPC_CQ_NEXT;
+  attr.cq_polling_type = GRPC_CQ_DEFAULT_POLLING;
+  cq1 = grpc_completion_queue_create(
+      grpc_completion_queue_factory_lookup(&attr), &attr, NULL);
+
+  attr.cq_polling_type = GRPC_CQ_NON_LISTENING;
+  cq2 = grpc_completion_queue_create(
+      grpc_completion_queue_factory_lookup(&attr), &attr, NULL);
+
+  attr.cq_polling_type = GRPC_CQ_NON_POLLING;
+  cq3 = grpc_completion_queue_create(
+      grpc_completion_queue_factory_lookup(&attr), &attr, NULL);
+
   server = grpc_server_create(NULL, NULL);
   grpc_server_register_completion_queue(server, cq1, NULL);
   grpc_server_add_insecure_http2_port(server, "[::]:0");
   grpc_server_register_completion_queue(server, cq2, NULL);
+  grpc_server_register_completion_queue(server, cq3, NULL);
+
   grpc_server_start(server);
   grpc_server_shutdown_and_notify(server, cq2, NULL);
   grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME),
                              NULL); /* cue queue hang */
   grpc_completion_queue_shutdown(cq1);
   grpc_completion_queue_shutdown(cq2);
+  grpc_completion_queue_shutdown(cq3);
+
   grpc_completion_queue_next(cq1, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
   grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+  grpc_completion_queue_next(cq3, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+
   grpc_server_destroy(server);
   grpc_completion_queue_destroy(cq1);
   grpc_completion_queue_destroy(cq2);
+  grpc_completion_queue_destroy(cq3);
   grpc_shutdown();
   return 0;
 }

test/core/end2end/no_server_test.c

@@ -59,7 +59,7 @@ int main(int argc, char **argv) {
 
   grpc_metadata_array_init(&trailing_metadata_recv);
 
-  cq = grpc_completion_queue_create(NULL);
+  cq = grpc_completion_queue_create_for_next(NULL);
   cqv = cq_verifier_create(cq);
 
   /* create a call, channel to a non existant server */

test/core/end2end/tests/authority_not_supported.c

@@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Request/response with metadata and payload.*/

test/core/end2end/tests/bad_hostname.c

@@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_fixture f) {

test/core/end2end/tests/bad_ping.c

@@ -75,6 +75,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void test_bad_ping(grpc_end2end_test_config config) {

test/core/end2end/tests/binary_metadata.c

@@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Request/response with metadata and payload.*/

test/core/end2end/tests/call_creds.c

@@ -92,9 +92,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -113,6 +114,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void print_auth_context(int is_client, const grpc_auth_context *ctx) {

test/core/end2end/tests/cancel_after_accept.c

@@ -81,9 +81,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -102,6 +103,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Cancel after accept, no payload */

test/core/end2end/tests/cancel_after_client_done.c

@@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Cancel after accept with a writes closed, no payload */

test/core/end2end/tests/cancel_after_invoke.c

@@ -78,9 +78,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
   grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
-                 .type == GRPC_OP_COMPLETE);
+  grpc_event ev = grpc_completion_queue_next(
+      f->cq, grpc_timeout_seconds_to_deadline(5), NULL);
+  GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+  GPR_ASSERT(ev.tag == tag(1000));
   grpc_server_destroy(f->server);
   f->server = NULL;
 }
@@ -98,6 +99,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Cancel after invoke, no payload */

test/core/end2end/tests/cancel_before_invoke.c

@@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Cancel before invoke */

test/core/end2end/tests/cancel_in_a_vacuum.c

@@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Cancel and do nothing */

test/core/end2end/tests/cancel_with_status.c

@@ -77,9 +77,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
   grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
-                 .type == GRPC_OP_COMPLETE);
+  grpc_event ev = grpc_completion_queue_next(
+      f->cq, grpc_timeout_seconds_to_deadline(5), NULL);
+  GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+  GPR_ASSERT(ev.tag == tag(1000));
   grpc_server_destroy(f->server);
   f->server = NULL;
 }
@@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_config config,

test/core/end2end/tests/compressed_payload.c

@@ -82,9 +82,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -103,6 +104,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void request_for_disabled_algorithm(

test/core/end2end/tests/connectivity.c

@@ -171,6 +171,9 @@ static void test_connectivity(grpc_end2end_test_config config) {
   grpc_channel_destroy(f.client);
   grpc_completion_queue_shutdown(f.cq);
   grpc_completion_queue_destroy(f.cq);
+
+  /* shutdown_cq is not used in this test */
+  grpc_completion_queue_destroy(f.shutdown_cq);
   config.tear_down_data(&f);
 
   cq_verifier_destroy(cqv);

test/core/end2end/tests/default_host.c

@@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_fixture f) {

test/core/end2end/tests/disappearing_server.c

@@ -79,6 +79,9 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+
+  /* Note: shutdown_cq was unused in this test */
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void do_request_and_shutdown_server(grpc_end2end_test_config config,

test/core/end2end/tests/empty_batch.c

@@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void empty_batch_body(grpc_end2end_test_config config,

test/core/end2end/tests/filter_call_init_fails.c

@@ -84,9 +84,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -105,6 +106,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 // Simple request via a SERVER_CHANNEL filter that always fails to

test/core/end2end/tests/filter_causes_close.c

@@ -79,9 +79,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -100,6 +101,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Simple request via a server filter that always closes the stream.*/

test/core/end2end/tests/filter_latency.c

@@ -86,9 +86,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -107,6 +108,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 // Simple request via a server filter that saves the reported latency value.

test/core/end2end/tests/graceful_server_shutdown.c

@@ -91,6 +91,8 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  /* Note: shutdown_cq is not used in this test */
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void test_early_server_shutdown_finishes_inflight_calls(

test/core/end2end/tests/high_initial_seqno.c

@@ -78,9 +78,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -99,6 +100,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_config config,

test/core/end2end/tests/hpack_size.c

@@ -218,9 +218,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -239,6 +240,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_config config,

test/core/end2end/tests/idempotent_request.c

@@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_config config,

test/core/end2end/tests/invoke_large_request.c

@@ -71,9 +71,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -92,6 +93,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static grpc_slice large_slice(void) {

test/core/end2end/tests/keepalive_timeout.c

@@ -78,8 +78,9 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(1000),
+
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
                                          five_seconds_from_now(), NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
@@ -99,6 +100,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 /* Client sends a request, server replies with a payload, then waits for the

test/core/end2end/tests/large_metadata.c

@@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 // Request with a large amount of metadata.

test/core/end2end/tests/load_reporting_hook.c

@@ -101,9 +101,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -122,6 +123,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void request_response_with_payload(

test/core/end2end/tests/max_concurrent_streams.c

@@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(
-                 f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
+  grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
+                                         grpc_timeout_seconds_to_deadline(5),
+                                         NULL)
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
@@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_shutdown(f->cq);
   drain_cq(f->cq);
   grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
 }
 
 static void simple_request_body(grpc_end2end_test_config config,