Merge branch 'bye-bye-completion-queue-pie' into we-dont-need-no-backup

37 files changed, 215 insertions, 540 deletions

diff --git a/test/core/end2end/cq_verifier.c b/test/core/end2end/cq_verifier.c
index f291e73e3b..66480254d2 100644
--- a/test/core/end2end/cq_verifier.c
+++ b/test/core/end2end/cq_verifier.c
@@ -45,6 +45,8 @@
 #include <grpc/support/time.h>
 #include <grpc/support/useful.h>
 
+#define ROOT_EXPECTATION 1000
+
 /* a set of metadata we expect to find on an event */
 typedef struct metadata {
   size_t count;
@@ -60,9 +62,7 @@ typedef struct expectation {
   struct expectation *prev;
   grpc_completion_type type;
   void *tag;
-  union {
-    grpc_op_error op_complete;
-  } data;
+  int success;
 } expectation;
 
 /* the verifier itself */
@@ -75,7 +75,7 @@ struct cq_verifier {
 
 cq_verifier *cq_verifier_create(grpc_completion_queue *cq) {
   cq_verifier *v = gpr_malloc(sizeof(cq_verifier));
-  v->expect.type = GRPC_COMPLETION_DO_NOT_USE;
+  v->expect.type = ROOT_EXPECTATION;
   v->expect.tag = NULL;
   v->expect.next = &v->expect;
   v->expect.prev = &v->expect;
@@ -149,11 +149,9 @@ static void verify_matches(expectation *e, grpc_event *ev) {
       abort();
       break;
     case GRPC_OP_COMPLETE:
-      GPR_ASSERT(e->data.op_complete == ev->data.op_complete);
-      break;
-    case GRPC_SERVER_SHUTDOWN:
+      GPR_ASSERT(e->success == ev->success);
       break;
-    case GRPC_COMPLETION_DO_NOT_USE:
+    case GRPC_QUEUE_TIMEOUT:
       gpr_log(GPR_ERROR, "not implemented");
       abort();
       break;
@@ -165,13 +163,10 @@ static void expectation_to_strvec(gpr_strvec *buf, expectation *e) {
 
   switch (e->type) {
     case GRPC_OP_COMPLETE:
-      gpr_asprintf(&tmp, "GRPC_OP_COMPLETE result=%d", e->data.op_complete);
+      gpr_asprintf(&tmp, "GRPC_OP_COMPLETE result=%d", e->success);
       gpr_strvec_add(buf, tmp);
       break;
-    case GRPC_SERVER_SHUTDOWN:
-      gpr_strvec_add(buf, gpr_strdup("GRPC_SERVER_SHUTDOWN"));
-      break;
-    case GRPC_COMPLETION_DO_NOT_USE:
+    case GRPC_QUEUE_TIMEOUT:
     case GRPC_QUEUE_SHUTDOWN:
       gpr_log(GPR_ERROR, "not implemented");
       abort();
@@ -203,7 +198,7 @@ static void fail_no_event_received(cq_verifier *v) {
 
 void cq_verify(cq_verifier *v) {
   gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);
-  grpc_event *ev;
+  grpc_event ev;
   expectation *e;
   char *s;
   gpr_strvec have_tags;
@@ -212,15 +207,16 @@ void cq_verify(cq_verifier *v) {
 
   while (v->expect.next != &v->expect) {
     ev = grpc_completion_queue_next(v->cq, deadline);
-    if (!ev) {
+    if (ev.type == GRPC_QUEUE_TIMEOUT) {
       fail_no_event_received(v);
+      break;
     }
 
     for (e = v->expect.next; e != &v->expect; e = e->next) {
       gpr_asprintf(&s, " %p", e->tag);
       gpr_strvec_add(&have_tags, s);
-      if (e->tag == ev->tag) {
-        verify_matches(e, ev);
+      if (e->tag == ev.tag) {
+        verify_matches(e, &ev);
         e->next->prev = e->prev;
         e->prev->next = e->next;
         gpr_free(e);
@@ -228,7 +224,7 @@ void cq_verify(cq_verifier *v) {
       }
     }
     if (e == &v->expect) {
-      s = grpc_event_string(ev);
+      s = grpc_event_string(&ev);
       gpr_log(GPR_ERROR, "event not found: %s", s);
       gpr_free(s);
       s = gpr_strvec_flatten(&have_tags, NULL);
@@ -237,8 +233,6 @@ void cq_verify(cq_verifier *v) {
       gpr_strvec_destroy(&have_tags);
       abort();
     }
-
-    grpc_event_finish(ev);
   }
 
   gpr_strvec_destroy(&have_tags);
@@ -246,13 +240,13 @@ void cq_verify(cq_verifier *v) {
 
 void cq_verify_empty(cq_verifier *v) {
   gpr_timespec deadline = gpr_time_add(gpr_now(), gpr_time_from_seconds(1));
-  grpc_event *ev;
+  grpc_event ev;
 
   GPR_ASSERT(v->expect.next == &v->expect && "expectation queue must be empty");
 
   ev = grpc_completion_queue_next(v->cq, deadline);
-  if (ev != NULL) {
-    char *s = grpc_event_string(ev);
+  if (ev.type != GRPC_QUEUE_TIMEOUT) {
+    char *s = grpc_event_string(&ev);
     gpr_log(GPR_ERROR, "unexpected event (expected nothing): %s", s);
     gpr_free(s);
     abort();
@@ -269,10 +263,6 @@ static expectation *add(cq_verifier *v, grpc_completion_type type, void *tag) {
   return e;
 }
 
-void cq_expect_completion(cq_verifier *v, void *tag, grpc_op_error result) {
-  add(v, GRPC_OP_COMPLETE, tag)->data.op_complete = result;
-}
-
-void cq_expect_server_shutdown(cq_verifier *v, void *tag) {
-  add(v, GRPC_SERVER_SHUTDOWN, tag);
+void cq_expect_completion(cq_verifier *v, void *tag, int success) {
+  add(v, GRPC_OP_COMPLETE, tag)->success = success;
 }
diff --git a/test/core/end2end/cq_verifier.h b/test/core/end2end/cq_verifier.h
index bae3c6caf0..1ecd4db5da 100644
--- a/test/core/end2end/cq_verifier.h
+++ b/test/core/end2end/cq_verifier.h
@@ -57,8 +57,7 @@ void cq_verify_empty(cq_verifier *v);
    Any functions taking ... expect a NULL terminated list of key/value pairs
    (each pair using two parameter slots) of metadata that MUST be present in
    the event. */
-void cq_expect_completion(cq_verifier *v, void *tag, grpc_op_error result);
-void cq_expect_server_shutdown(cq_verifier *v, void *tag);
+void cq_expect_completion(cq_verifier *v, void *tag, int success);
 
 int byte_buffer_eq_string(grpc_byte_buffer *byte_buffer, const char *string);
 int contains_metadata(grpc_metadata_array *array, const char *key, const char *value);
diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c
index d418ea8732..b54b8ecb3c 100644
--- a/test/core/end2end/dualstack_socket_test.c
+++ b/test/core/end2end/dualstack_socket_test.c
@@ -50,15 +50,10 @@ static gpr_timespec ms_from_now(int ms) {
 }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, ms_from_now(5000));
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-    gpr_log(GPR_INFO, "Drained event type %d", type);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 void test_connect(const char *server_host, const char *client_host, int port,
@@ -156,7 +151,7 @@ void test_connect(const char *server_host, const char *client_host, int port,
                grpc_server_request_call(server, &s, &call_details,
                                         &request_metadata_recv, cq,
                                         cq, tag(101)));
-    cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+    cq_expect_completion(cqv, tag(101), 1);
     cq_verify(cqv);
 
     op = ops;
@@ -174,8 +169,8 @@ void test_connect(const char *server_host, const char *client_host, int port,
     GPR_ASSERT(GRPC_CALL_OK ==
                grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-    cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-    cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+    cq_expect_completion(cqv, tag(102), 1);
+    cq_expect_completion(cqv, tag(1), 1);
     cq_verify(cqv);
 
     GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
@@ -187,7 +182,7 @@ void test_connect(const char *server_host, const char *client_host, int port,
     grpc_call_destroy(s);
   } else {
     /* Check for a failed connection. */
-    cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+    cq_expect_completion(cqv, tag(1), 1);
     cq_verify(cqv);
 
     GPR_ASSERT(status == GRPC_STATUS_DEADLINE_EXCEEDED);
diff --git a/test/core/end2end/no_server_test.c b/test/core/end2end/no_server_test.c
index b292620fe9..bba9cd1a56 100644
--- a/test/core/end2end/no_server_test.c
+++ b/test/core/end2end/no_server_test.c
@@ -45,8 +45,6 @@ int main(int argc, char **argv) {
   gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(2);
   grpc_completion_queue *cq;
   cq_verifier *cqv;
-  grpc_event *ev;
-  int done;
   grpc_op ops[6];
   grpc_op *op;
   grpc_metadata_array trailing_metadata_recv;
@@ -79,17 +77,15 @@ int main(int argc, char **argv) {
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_call_start_batch(call, ops, op - ops, tag(1)));
   /* verify that all tags get completed */
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_DEADLINE_EXCEEDED);
 
   grpc_completion_queue_shutdown(cq);
-  for (done = 0; !done;) {
-    ev = grpc_completion_queue_next(cq, gpr_inf_future);
-    done = ev->type == GRPC_QUEUE_SHUTDOWN;
-    grpc_event_finish(ev);
-  }
+  while (grpc_completion_queue_next(cq, gpr_inf_future).type !=
+         GRPC_QUEUE_SHUTDOWN)
+    ;
   grpc_completion_queue_destroy(cq);
   grpc_call_destroy(call);
   grpc_channel_destroy(chan);
diff --git a/test/core/end2end/tests/bad_hostname.c b/test/core/end2end/tests/bad_hostname.c
index d9971a616e..726bada7bf 100644
--- a/test/core/end2end/tests/bad_hostname.c
+++ b/test/core/end2end/tests/bad_hostname.c
@@ -68,14 +68,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -140,7 +136,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
 
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNAUTHENTICATED);
diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c
index 8ae53b77b2..db6d0899fe 100644
--- a/test/core/end2end/tests/cancel_after_accept.c
+++ b/test/core/end2end/tests/cancel_after_accept.c
@@ -67,14 +67,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -160,7 +156,7 @@ static void test_cancel_after_accept(grpc_end2end_test_config config,
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(2)));
-  cq_expect_completion(cqv, tag(2), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(2), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -180,8 +176,8 @@ static void test_cancel_after_accept(grpc_end2end_test_config config,
 
   GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
 
-  cq_expect_completion(cqv, tag(3), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(3), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == mode.expect_status);
diff --git a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
index 4b7fdd62b3..0162df7d28 100644
--- a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
+++ b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
@@ -67,14 +67,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -162,7 +158,7 @@ static void test_cancel_after_accept_and_writes_closed(
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(2)));
-  cq_expect_completion(cqv, tag(2), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(2), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -182,8 +178,8 @@ static void test_cancel_after_accept_and_writes_closed(
 
   GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
 
-  cq_expect_completion(cqv, tag(3), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(3), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == mode.expect_status);
diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c
index 7ee4350e11..cf12269d00 100644
--- a/test/core/end2end/tests/cancel_after_invoke.c
+++ b/test/core/end2end/tests/cancel_after_invoke.c
@@ -68,14 +68,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -156,7 +152,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config,
 
   GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
 
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == mode.expect_status);
diff --git a/test/core/end2end/tests/cancel_before_invoke.c b/test/core/end2end/tests/cancel_before_invoke.c
index 8a53e18535..19d3badde1 100644
--- a/test/core/end2end/tests/cancel_before_invoke.c
+++ b/test/core/end2end/tests/cancel_before_invoke.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -153,7 +149,7 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config,
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, test_ops, tag(1)));
 
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
diff --git a/test/core/end2end/tests/cancel_in_a_vacuum.c b/test/core/end2end/tests/cancel_in_a_vacuum.c
index 467a621db9..679b137521 100644
--- a/test/core/end2end/tests/cancel_in_a_vacuum.c
+++ b/test/core/end2end/tests/cancel_in_a_vacuum.c
@@ -65,14 +65,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
diff --git a/test/core/end2end/tests/census_simple_request.c b/test/core/end2end/tests/census_simple_request.c
index 4f18eabab3..4f6c57c4ed 100644
--- a/test/core/end2end/tests/census_simple_request.c
+++ b/test/core/end2end/tests/census_simple_request.c
@@ -74,14 +74,10 @@ static void shutdown_client(grpc_end2end_test_fixture *f) {
 }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, n_seconds_time(5));
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void end_test(grpc_end2end_test_fixture *f) {
@@ -141,7 +137,7 @@ static void test_body(grpc_end2end_test_fixture f) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -158,8 +154,8 @@ static void test_body(grpc_end2end_test_fixture f) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c
index 1e5560f40f..9bcc79d34d 100644
--- a/test/core/end2end/tests/disappearing_server.c
+++ b/test/core/end2end/tests/disappearing_server.c
@@ -54,14 +54,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -132,7 +128,7 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f,
              grpc_server_request_call(f->server, &s, &call_details,
                                       &request_metadata_recv, f->cq,
                                       f->cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   /* should be able to shut down the server early
@@ -153,8 +149,8 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f,
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
index 64f2f61bba..3efbf6add8 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -147,7 +143,7 @@ static void test_early_server_shutdown_finishes_inflight_calls(
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -159,8 +155,8 @@ static void test_early_server_shutdown_finishes_inflight_calls(
   /* shutdown and destroy the server */
   shutdown_server(&f);
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
index b193820804..c52e54b135 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -116,7 +112,7 @@ static void test_early_server_shutdown_finishes_tags(
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
   grpc_server_shutdown(f.server);
-  cq_expect_completion(cqv, tag(101), GRPC_OP_ERROR);
+  cq_expect_completion(cqv, tag(101), 0);
   cq_verify(cqv);
   GPR_ASSERT(s == NULL);
 
diff --git a/test/core/end2end/tests/empty_batch.c b/test/core/end2end/tests/empty_batch.c
index b3f24aab8b..f1c3fb259f 100644
--- a/test/core/end2end/tests/empty_batch.c
+++ b/test/core/end2end/tests/empty_batch.c
@@ -68,14 +68,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -111,7 +107,7 @@ static void empty_batch_body(grpc_end2end_test_fixture f) {
   GPR_ASSERT(c);
 
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, op, 0, tag(1)));
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   grpc_call_destroy(c);
diff --git a/test/core/end2end/tests/graceful_server_shutdown.c b/test/core/end2end/tests/graceful_server_shutdown.c
index b5a3906dc5..70a06555fb 100644
--- a/test/core/end2end/tests/graceful_server_shutdown.c
+++ b/test/core/end2end/tests/graceful_server_shutdown.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -146,7 +142,7 @@ static void test_early_server_shutdown_finishes_inflight_calls(
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   /* shutdown and destroy the server */
@@ -167,12 +163,12 @@ static void test_early_server_shutdown_finishes_inflight_calls(
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
   cq_verify(cqv);
 
   grpc_call_destroy(s);
-  cq_expect_server_shutdown(cqv, tag(0xdead));
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(0xdead), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c
index 6fbd78d1a1..d4035d0674 100644
--- a/test/core/end2end/tests/invoke_large_request.c
+++ b/test/core/end2end/tests/invoke_large_request.c
@@ -64,14 +64,10 @@ static gpr_timespec n_seconds_time(int n) {
 }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, n_seconds_time(5));
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -164,7 +160,7 @@ static void test_invoke_large_request(grpc_end2end_test_config config) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -187,8 +183,8 @@ static void test_invoke_large_request(grpc_end2end_test_config config) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c
index 15b3c86f98..57300a0191 100644
--- a/test/core/end2end/tests/max_concurrent_streams.c
+++ b/test/core/end2end/tests/max_concurrent_streams.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -144,7 +140,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -161,8 +157,8 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
@@ -194,7 +190,7 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) {
   int live_call;
   gpr_timespec deadline;
   cq_verifier *cqv;
-  grpc_event *ev;
+  grpc_event ev;
   grpc_call_details call_details;
   grpc_metadata_array request_metadata_recv;
   grpc_metadata_array initial_metadata_recv1;
@@ -297,26 +293,26 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) {
 
   got_client_start = 0;
   got_server_start = 0;
+  live_call = -1;
   while (!got_client_start || !got_server_start) {
     ev = grpc_completion_queue_next(f.cq,
                                     GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3));
-    GPR_ASSERT(ev);
-    GPR_ASSERT(ev->type == GRPC_OP_COMPLETE);
-    GPR_ASSERT(ev->data.op_complete == GRPC_OP_OK);
-    if (ev->tag == tag(101)) {
+    GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+    GPR_ASSERT(ev.success);
+    if (ev.tag == tag(101)) {
       GPR_ASSERT(!got_server_start);
       got_server_start = 1;
     } else {
       GPR_ASSERT(!got_client_start);
-      GPR_ASSERT(ev->tag == tag(301) || ev->tag == tag(401));
+      GPR_ASSERT(ev.tag == tag(301) || ev.tag == tag(401));
       /* The /alpha or /beta calls started above could be invoked (but NOT both);
        * check this here */
       /* We'll get tag 303 or 403, we want 300, 400 */
-      live_call = ((int)(gpr_intptr)ev->tag) - 1;
+      live_call = ((int)(gpr_intptr)ev.tag) - 1;
       got_client_start = 1;
     }
-    grpc_event_finish(ev);
   }
+  GPR_ASSERT(live_call == 300 || live_call == 400);
 
   op = ops;
   op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -333,18 +329,18 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) {
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_call_start_batch(s1, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(live_call + 2), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(live_call + 2), 1);
   /* first request is finished, we should be able to start the second */
   live_call = (live_call == 300) ? 400 : 300;
-  cq_expect_completion(cqv, tag(live_call + 1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(live_call + 1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_server_request_call(f.server, &s2, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(201)));
-  cq_expect_completion(cqv, tag(201), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(201), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -362,8 +358,8 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) {
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_call_start_batch(s2, ops, op - ops, tag(202)));
 
-  cq_expect_completion(cqv, tag(live_call + 2), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(202), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(live_call + 2), 1);
+  cq_expect_completion(cqv, tag(202), 1);
   cq_verify(cqv);
 
   cq_verifier_destroy(cqv);
diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c
index d32930ca5a..364d59eea3 100644
--- a/test/core/end2end/tests/max_message_length.c
+++ b/test/core/end2end/tests/max_message_length.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -162,7 +158,7 @@ static void test_max_message_length(grpc_end2end_test_config config) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -171,8 +167,8 @@ static void test_max_message_length(grpc_end2end_test_config config) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
diff --git a/test/core/end2end/tests/no_op.c b/test/core/end2end/tests/no_op.c
index b7cd2f1e86..979eb06d6c 100644
--- a/test/core/end2end/tests/no_op.c
+++ b/test/core/end2end/tests/no_op.c
@@ -64,14 +64,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c
index bc9a2547e7..c00beb7770 100644
--- a/test/core/end2end/tests/ping_pong_streaming.c
+++ b/test/core/end2end/tests/ping_pong_streaming.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -152,7 +148,7 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(100)));
-  cq_expect_completion(cqv, tag(100), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(100), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -183,7 +179,7 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
     op++;
     GPR_ASSERT(GRPC_CALL_OK ==
                grpc_call_start_batch(s, ops, op - ops, tag(102)));
-    cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
+    cq_expect_completion(cqv, tag(102), 1);
     cq_verify(cqv);
 
     op = ops;
@@ -192,8 +188,8 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
     op++;
     GPR_ASSERT(GRPC_CALL_OK ==
                grpc_call_start_batch(s, ops, op - ops, tag(103)));
-    cq_expect_completion(cqv, tag(103), GRPC_OP_OK);
-    cq_expect_completion(cqv, tag(2), GRPC_OP_OK);
+    cq_expect_completion(cqv, tag(103), 1);
+    cq_expect_completion(cqv, tag(2), 1);
     cq_verify(cqv);
 
     grpc_byte_buffer_destroy(request_payload);
@@ -218,10 +214,10 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(104)));
 
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(3), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(104), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
+  cq_expect_completion(cqv, tag(3), 1);
+  cq_expect_completion(cqv, tag(101), 1);
+  cq_expect_completion(cqv, tag(104), 1);
   cq_verify(cqv);
 
   grpc_call_destroy(c);
diff --git a/test/core/end2end/tests/registered_call.c b/test/core/end2end/tests/registered_call.c
index b2bae66f6e..80f7d20dce 100644
--- a/test/core/end2end/tests/registered_call.c
+++ b/test/core/end2end/tests/registered_call.c
@@ -68,14 +68,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -145,7 +141,7 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -162,8 +158,8 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
index 58abf95592..ce872a4082 100644
--- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -180,7 +176,7 @@ static void test_request_response_with_metadata_and_payload(
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -204,8 +200,8 @@ static void test_request_response_with_metadata_and_payload(
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_response_with_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
index d5e5cb95b5..b4edb7362d 100644
--- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -166,7 +162,7 @@ static void test_request_response_with_metadata_and_payload(
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -190,8 +186,8 @@ static void test_request_response_with_metadata_and_payload(
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c
index 558b2dd049..cb50855561 100644
--- a/test/core/end2end/tests/request_response_with_payload.c
+++ b/test/core/end2end/tests/request_response_with_payload.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -158,7 +154,7 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -181,8 +177,8 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
index 3759bd0824..ec3f310eae 100644
--- a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
+++ b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
@@ -80,14 +80,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -214,11 +210,11 @@ static void request_response_with_payload_and_call_creds(
                                                       &request_metadata_recv,
                                                       f.cq, f.cq, 
                                                       tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   /* Cannot set creds on the server call object. */
-  GPR_ASSERT(grpc_call_set_credentials(s, NULL) != GRPC_CALL_OK);
+  GPR_ASSERT(!grpc_call_set_credentials(s, NULL));
 
   op = ops;
   op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -240,8 +236,8 @@ static void request_response_with_payload_and_call_creds(
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
index 9c0e1d214b..1f1cb4cb42 100644
--- a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -168,9 +164,14 @@ static void test_request_response_with_metadata_and_payload(
   GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
                                                       &call_details,
                                                       &request_metadata_recv,
+<<<<<<< HEAD
+                                                      f.server_cq, tag(101)));
+  cq_expect_completion(v_server, tag(101), 1);
+=======
                                                       f.server_cq, f.server_cq,
                                                       tag(101)));
   cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
+>>>>>>> a468c36601dd5997580129bbd66b5ebed02521f8
   cq_verify(v_server);
 
   op = ops;
@@ -195,10 +196,10 @@ static void test_request_response_with_metadata_and_payload(
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
+  cq_expect_completion(v_server, tag(102), 1);
   cq_verify(v_server);
 
-  cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
+  cq_expect_completion(v_client, tag(1), 1);
   cq_verify(v_client);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c
index 33228e18a0..a137786d50 100644
--- a/test/core/end2end/tests/request_with_large_metadata.c
+++ b/test/core/end2end/tests/request_with_large_metadata.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -162,7 +158,7 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -182,8 +178,8 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c
index 13e9d59203..86199c9f43 100644
--- a/test/core/end2end/tests/request_with_payload.c
+++ b/test/core/end2end/tests/request_with_payload.c
@@ -66,14 +66,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -153,7 +149,7 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -173,8 +169,8 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c
index 289acae9d6..c51019caff 100644
--- a/test/core/end2end/tests/simple_delayed_request.c
+++ b/test/core/end2end/tests/simple_delayed_request.c
@@ -54,14 +54,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -140,7 +136,7 @@ static void simple_delayed_request_body(grpc_end2end_test_config config,
              grpc_server_request_call(f->server, &s, &call_details,
                                       &request_metadata_recv, f->cq,
                                       f->cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -157,8 +153,8 @@ static void simple_delayed_request_body(grpc_end2end_test_config config,
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c
index 8dd5a2b266..be9e414b7c 100644
--- a/test/core/end2end/tests/simple_request.c
+++ b/test/core/end2end/tests/simple_request.c
@@ -68,14 +68,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -146,7 +142,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -163,8 +159,8 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
index 241da3f103..566f971686 100644
--- a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
+++ b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
@@ -68,14 +68,10 @@ static gpr_timespec n_seconds_time(int n) {
 static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
 
 static void drain_cq(grpc_completion_queue *cq) {
-  grpc_event *ev;
-  grpc_completion_type type;
+  grpc_event ev;
   do {
     ev = grpc_completion_queue_next(cq, five_seconds_time());
-    GPR_ASSERT(ev);
-    type = ev->type;
-    grpc_event_finish(ev);
-  } while (type != GRPC_QUEUE_SHUTDOWN);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -146,7 +142,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.cq,
                                       f.cq, tag(101)));
-  cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(101), 1);
   cq_verify(cqv);
 
   op = ops;
@@ -163,8 +159,8 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
   op++;
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
-  cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(102), 1);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/fling/client.c b/test/core/fling/client.c
index 68164b1c5a..37d787c7c3 100644
--- a/test/core/fling/client.c
+++ b/test/core/fling/client.c
@@ -93,7 +93,7 @@ static void step_ping_pong_request(void) {
                                   "localhost", gpr_inf_future);
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_call_start_batch(call, ops, op - ops, (void *)1));
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+  grpc_completion_queue_next(cq, gpr_inf_future);
   grpc_call_destroy(call);
   grpc_byte_buffer_destroy(response_payload_recv);
   call = NULL;
@@ -106,7 +106,7 @@ static void init_ping_pong_stream(void) {
   stream_init_op.data.send_initial_metadata.count = 0;
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_call_start_batch(call, &stream_init_op, 1, (void *)1));
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+  grpc_completion_queue_next(cq, gpr_inf_future);
 
   grpc_metadata_array_init(&initial_metadata_recv);
 
@@ -119,7 +119,7 @@ static void init_ping_pong_stream(void) {
 static void step_ping_pong_stream(void) {
   GPR_ASSERT(GRPC_CALL_OK ==
              grpc_call_start_batch(call, stream_step_ops, 2, (void *)1));
-  grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+  grpc_completion_queue_next(cq, gpr_inf_future);
   grpc_byte_buffer_destroy(response_payload_recv);
 }
 
@@ -147,7 +147,6 @@ int main(int argc, char **argv) {
   char *fake_argv[1];
 
   int payload_size = 1;
-  int done;
   int secure = 0;
   char *target = "localhost:443";
   gpr_cmdline *cl;
@@ -209,12 +208,9 @@ int main(int argc, char **argv) {
 
   grpc_channel_destroy(channel);
   grpc_completion_queue_shutdown(cq);
-  done = 0;
-  while (!done) {
-    grpc_event *ev = grpc_completion_queue_next(cq, gpr_inf_future);
-    done = (ev->type == GRPC_QUEUE_SHUTDOWN);
-    grpc_event_finish(ev);
-  }
+  while (grpc_completion_queue_next(cq, gpr_inf_future).type !=
+         GRPC_QUEUE_SHUTDOWN)
+    ;
   grpc_completion_queue_destroy(cq);
   grpc_byte_buffer_destroy(the_buffer);
   gpr_slice_unref(slice);
diff --git a/test/core/fling/server.c b/test/core/fling/server.c
index 1f1f857bd9..48304ed8d7 100644
--- a/test/core/fling/server.c
+++ b/test/core/fling/server.c
@@ -174,7 +174,7 @@ static void start_send_status(void) {
 static void sigint_handler(int x) { _exit(0); }
 
 int main(int argc, char **argv) {
-  grpc_event *ev;
+  grpc_event ev;
   call_state *s;
   char *addr_buf = NULL;
   gpr_cmdline *cl;
@@ -239,9 +239,8 @@ int main(int argc, char **argv) {
     }
     ev = grpc_completion_queue_next(
         cq, gpr_time_add(gpr_now(), gpr_time_from_micros(1000000)));
-    if (!ev) continue;
-    s = ev->tag;
-    switch (ev->type) {
+    s = ev.tag;
+    switch (ev.type) {
       case GRPC_OP_COMPLETE:
         switch ((gpr_intptr)s) {
           case FLING_SERVER_NEW_REQUEST:
@@ -303,10 +302,9 @@ int main(int argc, char **argv) {
         GPR_ASSERT(shutdown_started);
         shutdown_finished = 1;
         break;
-      default:
-        GPR_ASSERT(0);
+      case GRPC_QUEUE_TIMEOUT:
+        break;
     }
-    grpc_event_finish(ev);
   }
   grpc_profiler_stop();
   grpc_call_details_destroy(&call_details);
diff --git a/test/core/surface/completion_queue_benchmark.c b/test/core/surface/completion_queue_benchmark.c
deleted file mode 100644
index 81ebe15415..0000000000
--- a/test/core/surface/completion_queue_benchmark.c
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/core/surface/completion_queue.h"
-
-#include <math.h>
-#include <stdio.h>
-
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-#include <grpc/support/thd.h>
-#include <grpc/support/time.h>
-
-typedef struct test_thread_options {
-  gpr_event on_started;
-  gpr_event *start;
-  gpr_event on_finished;
-  grpc_completion_queue *cc;
-  int iterations;
-} test_thread_options;
-
-static void producer_thread(void *arg) {
-  test_thread_options *opt = arg;
-  int i;
-
-  gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1);
-  GPR_ASSERT(gpr_event_wait(opt->start, gpr_inf_future));
-
-  for (i = 0; i < opt->iterations; i++) {
-    grpc_cq_begin_op(opt->cc, NULL, GRPC_WRITE_ACCEPTED);
-    grpc_cq_end_write_accepted(opt->cc, (void *)(gpr_intptr) 1, NULL, NULL,
-                               NULL, GRPC_OP_OK);
-  }
-
-  gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1);
-}
-
-static void consumer_thread(void *arg) {
-  test_thread_options *opt = arg;
-  grpc_event *ev;
-
-  gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1);
-  GPR_ASSERT(gpr_event_wait(opt->start, gpr_inf_future));
-
-  for (;;) {
-    ev = grpc_completion_queue_next(opt->cc, gpr_inf_future);
-    switch (ev->type) {
-      case GRPC_WRITE_ACCEPTED:
-        break;
-      case GRPC_QUEUE_SHUTDOWN:
-        gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1);
-        return;
-      default:
-        gpr_log(GPR_ERROR, "Invalid event received: %d", ev->type);
-        abort();
-    }
-    grpc_event_finish(ev);
-  }
-}
-
-double ops_per_second(int consumers, int producers, int iterations) {
-  test_thread_options *options =
-      gpr_malloc((producers + consumers) * sizeof(test_thread_options));
-  gpr_event start = GPR_EVENT_INIT;
-  grpc_completion_queue *cc = grpc_completion_queue_create();
-  int i;
-  gpr_timespec t_start, t_end, t_delta;
-
-  /* start all threads: they will wait for phase1 */
-  for (i = 0; i < producers + consumers; i++) {
-    gpr_thd_id id;
-    gpr_event_init(&options[i].on_started);
-    gpr_event_init(&options[i].on_finished);
-    options[i].start = &start;
-    options[i].cc = cc;
-    options[i].iterations = iterations;
-    GPR_ASSERT(gpr_thd_new(&id,
-                           i < producers ? producer_thread : consumer_thread,
-                           options + i, NULL));
-    gpr_event_wait(&options[i].on_started, gpr_inf_future);
-  }
-
-  /* start the benchmark */
-  t_start = gpr_now();
-  gpr_event_set(&start, (void *)(gpr_intptr) 1);
-
-  /* wait for producers to finish */
-  for (i = 0; i < producers; i++) {
-    GPR_ASSERT(gpr_event_wait(&options[i].on_finished, gpr_inf_future));
-  }
-
-  /* in parallel, we shutdown the completion channel - all events should still
-     be consumed */
-  grpc_completion_queue_shutdown(cc);
-
-  /* join all threads */
-  for (i = producers; i < producers + consumers; i++) {
-    GPR_ASSERT(gpr_event_wait(&options[i].on_finished, gpr_inf_future));
-  }
-  t_end = gpr_now();
-
-  /* destroy the completion channel */
-  grpc_completion_queue_destroy(cc);
-
-  gpr_free(options);
-
-  t_delta = gpr_time_sub(t_end, t_start);
-  return (t_delta.tv_sec + 1e-9 * t_delta.tv_nsec) / (producers * iterations);
-}
-
-double ops_per_second_top(int consumers, int producers) {
-  return ops_per_second(consumers, producers, 1000000 / producers);
-}
-
-int main(void) {
-  const int counts[] = {1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 40, 64};
-  const int ncounts = sizeof(counts) / sizeof(*counts);
-  int i, j;
-
-  printf("\"\",");
-  for (i = 0; i < ncounts; i++) {
-    int producers = counts[i];
-    printf("%d%s", producers, i == ncounts - 1 ? "\n" : ",");
-  }
-
-  for (j = 0; j < ncounts; j++) {
-    int consumers = counts[j];
-    printf("%d,", consumers);
-    for (i = 0; i < ncounts; i++) {
-      int producers = counts[i];
-      printf("%f%s", ops_per_second_top(consumers, producers),
-             i == ncounts - 1 ? "\n" : ",");
-      fflush(stdout);
-    }
-  }
-
-  return 0;
-}
diff --git a/test/core/surface/completion_queue_test.c b/test/core/surface/completion_queue_test.c
index 29fb7a99a5..3e84eaf54c 100644
--- a/test/core/surface/completion_queue_test.c
+++ b/test/core/surface/completion_queue_test.c
@@ -43,10 +43,6 @@
 
 #define LOG_TEST() gpr_log(GPR_INFO, "%s", __FUNCTION__)
 
-static void increment_int_on_finish(void *user_data, grpc_op_error error) {
-  ++*(int *)user_data;
-}
-
 static void *create_test_tag(void) {
   static gpr_intptr i = 0;
   return (void *)(++i);
@@ -54,12 +50,10 @@ static void *create_test_tag(void) {
 
 /* helper for tests to shutdown correctly and tersely */
 static void shutdown_and_destroy(grpc_completion_queue *cc) {
-  grpc_event *ev;
+  grpc_event ev;
   grpc_completion_queue_shutdown(cc);
   ev = grpc_completion_queue_next(cc, gpr_inf_past);
-  GPR_ASSERT(ev != NULL);
-  GPR_ASSERT(ev->type == GRPC_QUEUE_SHUTDOWN);
-  grpc_event_finish(ev);
+  GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
   grpc_completion_queue_destroy(cc);
 }
 
@@ -75,42 +69,36 @@ static void test_wait_empty(void) {
   LOG_TEST();
 
   cc = grpc_completion_queue_create();
-  GPR_ASSERT(grpc_completion_queue_next(cc, gpr_now()) == NULL);
+  GPR_ASSERT(grpc_completion_queue_next(cc, gpr_now()).type ==
+             GRPC_QUEUE_TIMEOUT);
   shutdown_and_destroy(cc);
 }
 
 static void test_cq_end_op(void) {
-  grpc_event *ev;
+  grpc_event ev;
   grpc_completion_queue *cc;
-  int on_finish_called = 0;
   void *tag = create_test_tag();
 
   LOG_TEST();
 
   cc = grpc_completion_queue_create();
 
-  grpc_cq_begin_op(cc, NULL, GRPC_OP_COMPLETE);
-  grpc_cq_end_op(cc, tag, NULL, increment_int_on_finish, &on_finish_called,
-                 GRPC_OP_OK);
+  grpc_cq_begin_op(cc, NULL);
+  grpc_cq_end_op(cc, tag, NULL, 1);
 
   ev = grpc_completion_queue_next(cc, gpr_inf_past);
-  GPR_ASSERT(ev != NULL);
-  GPR_ASSERT(ev->type == GRPC_OP_COMPLETE);
-  GPR_ASSERT(ev->tag == tag);
-  GPR_ASSERT(ev->data.op_complete == GRPC_OP_OK);
-  GPR_ASSERT(on_finish_called == 0);
-  grpc_event_finish(ev);
-  GPR_ASSERT(on_finish_called == 1);
+  GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+  GPR_ASSERT(ev.tag == tag);
+  GPR_ASSERT(ev.success);
 
   shutdown_and_destroy(cc);
 }
 
 static void test_pluck(void) {
-  grpc_event *ev;
+  grpc_event ev;
   grpc_completion_queue *cc;
   void *tags[128];
   unsigned i, j;
-  int on_finish_called = 0;
 
   LOG_TEST();
 
@@ -124,34 +112,26 @@ static void test_pluck(void) {
   cc = grpc_completion_queue_create();
 
   for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
-    grpc_cq_begin_op(cc, NULL, GRPC_OP_COMPLETE);
-    grpc_cq_end_op(cc, tags[i], NULL, increment_int_on_finish,
-                   &on_finish_called, GRPC_OP_OK);
+    grpc_cq_begin_op(cc, NULL);
+    grpc_cq_end_op(cc, tags[i], NULL, 1);
   }
 
   for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
     ev = grpc_completion_queue_pluck(cc, tags[i], gpr_inf_past);
-    GPR_ASSERT(ev->tag == tags[i]);
-    grpc_event_finish(ev);
+    GPR_ASSERT(ev.tag == tags[i]);
   }
 
-  GPR_ASSERT(on_finish_called == GPR_ARRAY_SIZE(tags));
-
   for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
-    grpc_cq_begin_op(cc, NULL, GRPC_OP_COMPLETE);
-    grpc_cq_end_op(cc, tags[i], NULL, increment_int_on_finish,
-                   &on_finish_called, GRPC_OP_OK);
+    grpc_cq_begin_op(cc, NULL);
+    grpc_cq_end_op(cc, tags[i], NULL, 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_ASSERT(ev->tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
-    grpc_event_finish(ev);
+    GPR_ASSERT(ev.tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
   }
 
-  GPR_ASSERT(on_finish_called == 2 * GPR_ARRAY_SIZE(tags));
-
   shutdown_and_destroy(cc);
 }
 
@@ -182,7 +162,7 @@ static void producer_thread(void *arg) {
 
   gpr_log(GPR_INFO, "producer %d phase 1", opt->id);
   for (i = 0; i < TEST_THREAD_EVENTS; i++) {
-    grpc_cq_begin_op(opt->cc, NULL, GRPC_OP_COMPLETE);
+    grpc_cq_begin_op(opt->cc, NULL);
   }
 
   gpr_log(GPR_INFO, "producer %d phase 1 done", opt->id);
@@ -191,8 +171,7 @@ static void producer_thread(void *arg) {
 
   gpr_log(GPR_INFO, "producer %d phase 2", opt->id);
   for (i = 0; i < TEST_THREAD_EVENTS; i++) {
-    grpc_cq_end_op(opt->cc, (void *)(gpr_intptr)1, NULL, NULL, NULL,
-                   GRPC_OP_OK);
+    grpc_cq_end_op(opt->cc, (void *)(gpr_intptr)1, NULL, 1);
     opt->events_triggered++;
   }
 
@@ -202,7 +181,7 @@ static void producer_thread(void *arg) {
 
 static void consumer_thread(void *arg) {
   test_thread_options *opt = arg;
-  grpc_event *ev;
+  grpc_event ev;
 
   gpr_log(GPR_INFO, "consumer %d started", opt->id);
   gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1);
@@ -217,20 +196,17 @@ static void consumer_thread(void *arg) {
   gpr_log(GPR_INFO, "consumer %d phase 2", opt->id);
   for (;;) {
     ev = grpc_completion_queue_next(opt->cc, ten_seconds_time());
-    GPR_ASSERT(ev);
-    switch (ev->type) {
+    switch (ev.type) {
       case GRPC_OP_COMPLETE:
-        GPR_ASSERT(ev->data.op_complete == GRPC_OP_OK);
+        GPR_ASSERT(ev.success);
         opt->events_triggered++;
-        grpc_event_finish(ev);
         break;
       case GRPC_QUEUE_SHUTDOWN:
         gpr_log(GPR_INFO, "consumer %d phase 2 done", opt->id);
         gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1);
-        grpc_event_finish(ev);
         return;
-      default:
-        gpr_log(GPR_ERROR, "Invalid event received: %d", ev->type);
+      case GRPC_QUEUE_TIMEOUT:
+        gpr_log(GPR_ERROR, "Invalid timeout received");
         abort();
     }
   }
diff --git a/test/core/surface/lame_client_test.c b/test/core/surface/lame_client_test.c
index 05e8f95d71..34d37f0f3c 100644
--- a/test/core/surface/lame_client_test.c
+++ b/test/core/surface/lame_client_test.c
@@ -79,7 +79,7 @@ int main(int argc, char **argv) {
              grpc_call_start_batch(call, ops, op - ops, tag(1)));
 
   /* the call should immediately fail */
-  cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+  cq_expect_completion(cqv, tag(1), 1);
   cq_verify(cqv);
 
   grpc_call_destroy(call);