Merge pull request #10662 from sreecha/cq_mpsc_based

grpc_mpsc queue based completion queue

build.yaml

@@ -3343,7 +3343,7 @@ targets:
   - gpr_test_util
   - gpr
   args:
-  - --benchmark_min_time=0
+  - --benchmark_min_time=4
   defaults: benchmark
   platforms:
   - mac

include/grpc/impl/codegen/grpc_types.h

@@ -589,7 +589,7 @@ typedef enum {
 /** 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 */
-  GRPC_CQ_NEXT = 1,
+  GRPC_CQ_NEXT,
 
   /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
   GRPC_CQ_PLUCK

src/core/lib/surface/completion_queue.h

@@ -50,6 +50,8 @@ extern grpc_tracer_flag grpc_trace_pending_tags;
 #endif
 
 typedef struct grpc_cq_completion {
+  gpr_mpscq_node node;
+
   /** user supplied tag */
   void *tag;
   /** done callback - called when this queue element is no longer

test/cpp/microbenchmarks/bm_cq_multiple_threads.cc

@@ -81,10 +81,16 @@ static void cq_done_cb(grpc_exec_ctx* exec_ctx, void* done_arg,
   gpr_free(cq_completion);
 }
 
-/* Queues a completion tag. ZERO polling overhead */
+/* Queues a completion tag if deadline is > 0.
+ * Does nothing if deadline is 0 (i.e gpr_time_0(GPR_CLOCK_MONOTONIC)) */
 static grpc_error* pollset_work(grpc_exec_ctx* exec_ctx, grpc_pollset* ps,
                                 grpc_pollset_worker** worker, gpr_timespec now,
                                 gpr_timespec deadline) {
+  if (gpr_time_cmp(deadline, gpr_time_0(GPR_CLOCK_MONOTONIC)) == 0) {
+    gpr_log(GPR_ERROR, "no-op");
+    return GRPC_ERROR_NONE;
+  }
+
   gpr_mu_unlock(&ps->mu);
   grpc_cq_begin_op(g_cq, g_tag);
   grpc_cq_end_op(exec_ctx, g_cq, g_tag, GRPC_ERROR_NONE, cq_done_cb, NULL,
@@ -115,6 +121,14 @@ static void setup() {
 
 static void teardown() {
   grpc_completion_queue_shutdown(g_cq);
+
+  /* Drain any events */
+  gpr_timespec deadline = gpr_time_0(GPR_CLOCK_MONOTONIC);
+  while (grpc_completion_queue_next(g_cq, deadline, NULL).type !=
+         GRPC_QUEUE_SHUTDOWN) {
+    /* Do nothing */
+  }
+
   grpc_completion_queue_destroy(g_cq);
 }
 

tools/run_tests/generated/tests.json

@@ -2805,7 +2805,7 @@
   }, 
   {
     "args": [
-      "--benchmark_min_time=0"
+      "--benchmark_min_time=4"
     ], 
     "ci_platforms": [
       "linux",