Change Core to use the new completion_queue_create API

1 files changed, 134 insertions, 58 deletions

diff --git a/test/core/surface/completion_queue_test.c b/test/core/surface/completion_queue_test.c
index 07f6a9869b..6d913cc932 100644
--- a/test/core/surface/completion_queue_test.c
+++ b/test/core/surface/completion_queue_test.c
@@ -51,24 +51,62 @@ static void *create_test_tag(void) {
 static void shutdown_and_destroy(grpc_completion_queue *cc) {
   grpc_event ev;
   grpc_completion_queue_shutdown(cc);
-  ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
+
+  switch (grpc_get_cq_completion_type(cc)) {
+    case GRPC_CQ_NEXT: {
+      ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                      NULL);
+      break;
+    }
+    case GRPC_CQ_PLUCK: {
+      ev = grpc_completion_queue_pluck(cc, create_test_tag(),
+                                       gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
+      break;
+    }
+    default: {
+      gpr_log(GPR_ERROR, "Unknown completion type");
+      break;
+    }
+  }
+
   GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
   grpc_completion_queue_destroy(cc);
 }
 
 /* ensure we can create and destroy a completion channel */
 static void test_no_op(void) {
+  grpc_cq_completion_type completion_types[] = {GRPC_CQ_NEXT, GRPC_CQ_PLUCK};
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
+                                          NON_POLLING};
   LOG_TEST("test_no_op");
-  shutdown_and_destroy(grpc_completion_queue_create(NULL));
+
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(completion_types); i++) {
+    for (size_t j = 0; j < GPR_ARRAY_SIZE(polling_types); j++) {
+      shutdown_and_destroy(grpc_completion_queue_create(
+          completion_types[i], polling_types[j], NULL));
+    }
+  }
 }
 
 static void test_pollset_conversion(void) {
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
-  GPR_ASSERT(grpc_cq_from_pollset(grpc_cq_pollset(cq)) == cq);
-  shutdown_and_destroy(cq);
+  grpc_cq_completion_type completion_types[] = {GRPC_CQ_NEXT, GRPC_CQ_PLUCK};
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING};
+  grpc_completion_queue *cq;
+
+  LOG_TEST("test_pollset_conversion");
+
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(completion_types); i++) {
+    for (size_t j = 0; j < GPR_ARRAY_SIZE(polling_types); j++) {
+      cq = grpc_completion_queue_create(completion_types[i], polling_types[j],
+                                        NULL);
+      GPR_ASSERT(grpc_cq_from_pollset(grpc_cq_pollset(cq)) == cq);
+      shutdown_and_destroy(cq);
+    }
+  }
 }
 
 static void test_wait_empty(void) {
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING};
   grpc_completion_queue *cc;
   grpc_event event;
 
@@ -87,50 +125,66 @@ static void test_cq_end_op(void) {
   grpc_event ev;
   grpc_completion_queue *cc;
   grpc_cq_completion completion;
-  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
+                                          NON_POLLING};
+
+  grpc_exec_ctx init_exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_exec_ctx exec_ctx;
   void *tag = create_test_tag();
 
   LOG_TEST("test_cq_end_op");
 
-  cc = grpc_completion_queue_create(NULL);
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
+    exec_ctx = init_exec_ctx;  // Reset exec_ctx
+    cc = grpc_completion_queue_create(GRPC_CQ_NEXT, polling_types[i], NULL);
 
-  grpc_cq_begin_op(cc, tag);
-  grpc_cq_end_op(&exec_ctx, cc, tag, GRPC_ERROR_NONE, do_nothing_end_completion,
-                 NULL, &completion);
+    grpc_cq_begin_op(cc, tag);
+    grpc_cq_end_op(&exec_ctx, cc, tag, GRPC_ERROR_NONE,
+                   do_nothing_end_completion, NULL, &completion);
 
-  ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
-  GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
-  GPR_ASSERT(ev.tag == tag);
-  GPR_ASSERT(ev.success);
+    ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
+    GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+    GPR_ASSERT(ev.tag == tag);
+    GPR_ASSERT(ev.success);
 
-  shutdown_and_destroy(cc);
-  grpc_exec_ctx_finish(&exec_ctx);
+    shutdown_and_destroy(cc);
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
 }
 
 static void test_shutdown_then_next_polling(void) {
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
+                                          NON_POLLING};
   grpc_completion_queue *cc;
   grpc_event event;
   LOG_TEST("test_shutdown_then_next_polling");
 
-  cc = grpc_completion_queue_create(NULL);
-  grpc_completion_queue_shutdown(cc);
-  event =
-      grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
-  GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN);
-  grpc_completion_queue_destroy(cc);
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
+    cc = grpc_completion_queue_create(GRPC_CQ_NEXT, polling_types[i], NULL);
+    grpc_completion_queue_shutdown(cc);
+    event =
+        grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
+    GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN);
+    grpc_completion_queue_destroy(cc);
+  }
 }
 
 static void test_shutdown_then_next_with_timeout(void) {
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
+                                          NON_POLLING};
   grpc_completion_queue *cc;
   grpc_event event;
   LOG_TEST("test_shutdown_then_next_with_timeout");
 
-  cc = grpc_completion_queue_create(NULL);
-  grpc_completion_queue_shutdown(cc);
-  event =
-      grpc_completion_queue_next(cc, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
-  GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN);
-  grpc_completion_queue_destroy(cc);
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
+    cc = grpc_completion_queue_create(GRPC_CQ_NEXT, polling_types[i], NULL);
+
+    grpc_completion_queue_shutdown(cc);
+    event = grpc_completion_queue_next(cc, gpr_inf_future(GPR_CLOCK_REALTIME),
+                                       NULL);
+    GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN);
+    grpc_completion_queue_destroy(cc);
+  }
 }
 
 static void test_pluck(void) {
@@ -138,7 +192,10 @@ static void test_pluck(void) {
   grpc_completion_queue *cc;
   void *tags[128];
   grpc_cq_completion completions[GPR_ARRAY_SIZE(tags)];
-  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
+                                          NON_POLLING};
+  grpc_exec_ctx init_exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_exec_ctx exec_ctx;
   unsigned i, j;
 
   LOG_TEST("test_pluck");
@@ -150,47 +207,66 @@ static void test_pluck(void) {
     }
   }
 
-  cc = grpc_completion_queue_create(NULL);
+  for (size_t pidx = 0; pidx < GPR_ARRAY_SIZE(polling_types); pidx++) {
+    exec_ctx = init_exec_ctx;  // reset exec_ctx
+    cc = grpc_completion_queue_create(GRPC_CQ_PLUCK, polling_types[pidx], NULL);
 
-  for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
-    grpc_cq_begin_op(cc, tags[i]);
-    grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
-                   do_nothing_end_completion, NULL, &completions[i]);
-  }
+    for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
+      grpc_cq_begin_op(cc, tags[i]);
+      grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
+                     do_nothing_end_completion, NULL, &completions[i]);
+    }
 
-  for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
-    ev = grpc_completion_queue_pluck(cc, tags[i],
-                                     gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
-    GPR_ASSERT(ev.tag == tags[i]);
-  }
+    for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
+      ev = grpc_completion_queue_pluck(cc, tags[i],
+                                       gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
+      GPR_ASSERT(ev.tag == tags[i]);
+    }
 
-  for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
-    grpc_cq_begin_op(cc, tags[i]);
-    grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
-                   do_nothing_end_completion, NULL, &completions[i]);
-  }
+    for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
+      grpc_cq_begin_op(cc, tags[i]);
+      grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
+                     do_nothing_end_completion, NULL, &completions[i]);
+    }
 
-  for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
-    ev = grpc_completion_queue_pluck(cc, tags[GPR_ARRAY_SIZE(tags) - i - 1],
-                                     gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
-    GPR_ASSERT(ev.tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
-  }
+    for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
+      ev = grpc_completion_queue_pluck(cc, tags[GPR_ARRAY_SIZE(tags) - i - 1],
+                                       gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
+      GPR_ASSERT(ev.tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
+    }
 
-  shutdown_and_destroy(cc);
-  grpc_exec_ctx_finish(&exec_ctx);
+    shutdown_and_destroy(cc);
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
 }
 
 static void test_pluck_after_shutdown(void) {
+  grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
+                                          NON_POLLING};
   grpc_event ev;
   grpc_completion_queue *cc;
 
   LOG_TEST("test_pluck_after_shutdown");
-  cc = grpc_completion_queue_create(NULL);
-  grpc_completion_queue_shutdown(cc);
-  ev = grpc_completion_queue_pluck(cc, NULL, gpr_inf_future(GPR_CLOCK_REALTIME),
-                                   NULL);
-  GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
-  grpc_completion_queue_destroy(cc);
+
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
+    cc = grpc_completion_queue_create(GRPC_CQ_PLUCK, polling_types[i], NULL);
+    grpc_completion_queue_shutdown(cc);
+    ev = grpc_completion_queue_pluck(cc, NULL,
+                                     gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+    GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
+    grpc_completion_queue_destroy(cc);
+  }
+}
+
+struct thread_state {
+  grpc_completion_queue *cc;
+  void *tag;
+};
+
+static void pluck_one(void *arg) {
+  struct thread_state *state = arg;
+  grpc_completion_queue_pluck(state->cc, state->tag,
+                              gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
 }
 
 int main(int argc, char **argv) {