Merge branch 'master' of https://github.com/grpc/grpc into ruby_advanced_interop_client_side

41 files changed, 416 insertions, 212 deletions

diff --git a/test/core/bad_client/bad_client.c b/test/core/bad_client/bad_client.c
index 07fcd995d7..d579dcc7d4 100644
--- a/test/core/bad_client/bad_client.c
+++ b/test/core/bad_client/bad_client.c
@@ -148,7 +148,8 @@ void grpc_run_bad_client_test(
 
   grpc_slice_buffer_init(&outgoing);
   grpc_slice_buffer_add(&outgoing, slice);
-  grpc_closure_init(&done_write_closure, done_write, &a);
+  grpc_closure_init(&done_write_closure, done_write, &a,
+                    grpc_schedule_on_exec_ctx);
 
   /* Write data */
   grpc_endpoint_write(&exec_ctx, sfd.client, &outgoing, &done_write_closure);
@@ -175,7 +176,8 @@ void grpc_run_bad_client_test(
       grpc_slice_buffer_init(&args.incoming);
       gpr_event_init(&args.read_done);
       grpc_closure read_done_closure;
-      grpc_closure_init(&read_done_closure, read_done, &args);
+      grpc_closure_init(&read_done_closure, read_done, &args,
+                        grpc_schedule_on_exec_ctx);
       grpc_endpoint_read(&exec_ctx, sfd.client, &args.incoming,
                          &read_done_closure);
       grpc_exec_ctx_finish(&exec_ctx);
diff --git a/test/core/client_channel/resolvers/dns_resolver_connectivity_test.c b/test/core/client_channel/resolvers/dns_resolver_connectivity_test.c
index b421720492..169323e0f7 100644
--- a/test/core/client_channel/resolvers/dns_resolver_connectivity_test.c
+++ b/test/core/client_channel/resolvers/dns_resolver_connectivity_test.c
@@ -108,16 +108,18 @@ int main(int argc, char **argv) {
   grpc_resolver *resolver = create_resolver(&exec_ctx, "dns:test");
   gpr_event ev1;
   gpr_event_init(&ev1);
-  grpc_resolver_next(&exec_ctx, resolver, &result,
-                     grpc_closure_create(on_done, &ev1));
+  grpc_resolver_next(
+      &exec_ctx, resolver, &result,
+      grpc_closure_create(on_done, &ev1, grpc_schedule_on_exec_ctx));
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(wait_loop(5, &ev1));
   GPR_ASSERT(result == NULL);
 
   gpr_event ev2;
   gpr_event_init(&ev2);
-  grpc_resolver_next(&exec_ctx, resolver, &result,
-                     grpc_closure_create(on_done, &ev2));
+  grpc_resolver_next(
+      &exec_ctx, resolver, &result,
+      grpc_closure_create(on_done, &ev2, grpc_schedule_on_exec_ctx));
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(wait_loop(30, &ev2));
   GPR_ASSERT(result != NULL);
diff --git a/test/core/client_channel/resolvers/sockaddr_resolver_test.c b/test/core/client_channel/resolvers/sockaddr_resolver_test.c
index a9fd85aea1..d6c8920ad0 100644
--- a/test/core/client_channel/resolvers/sockaddr_resolver_test.c
+++ b/test/core/client_channel/resolvers/sockaddr_resolver_test.c
@@ -68,8 +68,8 @@ static void test_succeeds(grpc_resolver_factory *factory, const char *string) {
   on_resolution_arg on_res_arg;
   memset(&on_res_arg, 0, sizeof(on_res_arg));
   on_res_arg.expected_server_name = uri->path;
-  grpc_closure *on_resolution =
-      grpc_closure_create(on_resolution_cb, &on_res_arg);
+  grpc_closure *on_resolution = grpc_closure_create(
+      on_resolution_cb, &on_res_arg, grpc_schedule_on_exec_ctx);
 
   grpc_resolver_next(&exec_ctx, resolver, &on_res_arg.resolver_result,
                      on_resolution);
diff --git a/test/core/client_channel/set_initial_connect_string_test.c b/test/core/client_channel/set_initial_connect_string_test.c
index 11e57439d5..2082f65458 100644
--- a/test/core/client_channel/set_initial_connect_string_test.c
+++ b/test/core/client_channel/set_initial_connect_string_test.c
@@ -94,7 +94,7 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp,
                        grpc_tcp_server_acceptor *acceptor) {
   gpr_free(acceptor);
   test_tcp_server *server = arg;
-  grpc_closure_init(&on_read, handle_read, NULL);
+  grpc_closure_init(&on_read, handle_read, NULL, grpc_schedule_on_exec_ctx);
   grpc_slice_buffer_init(&state.incoming_buffer);
   grpc_slice_buffer_init(&state.temp_incoming_buffer);
   state.tcp = tcp;
diff --git a/test/core/end2end/bad_server_response_test.c b/test/core/end2end/bad_server_response_test.c
index 30468558e8..f6a9cbeef9 100644
--- a/test/core/end2end/bad_server_response_test.c
+++ b/test/core/end2end/bad_server_response_test.c
@@ -147,8 +147,8 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp,
                        grpc_tcp_server_acceptor *acceptor) {
   gpr_free(acceptor);
   test_tcp_server *server = arg;
-  grpc_closure_init(&on_read, handle_read, NULL);
-  grpc_closure_init(&on_write, done_write, NULL);
+  grpc_closure_init(&on_read, handle_read, NULL, grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&on_write, done_write, NULL, grpc_schedule_on_exec_ctx);
   grpc_slice_buffer_init(&state.temp_incoming_buffer);
   grpc_slice_buffer_init(&state.outgoing_buffer);
   state.tcp = tcp;
diff --git a/test/core/end2end/fake_resolver.c b/test/core/end2end/fake_resolver.c
index ed85030797..45d48720c6 100644
--- a/test/core/end2end/fake_resolver.c
+++ b/test/core/end2end/fake_resolver.c
@@ -87,7 +87,7 @@ static void fake_resolver_shutdown(grpc_exec_ctx* exec_ctx,
   gpr_mu_lock(&r->mu);
   if (r->next_completion != NULL) {
     *r->target_result = NULL;
-    grpc_exec_ctx_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE);
     r->next_completion = NULL;
   }
   gpr_mu_unlock(&r->mu);
@@ -100,7 +100,7 @@ static void fake_resolver_maybe_finish_next_locked(grpc_exec_ctx* exec_ctx,
     grpc_arg arg = grpc_lb_addresses_create_channel_arg(r->addresses);
     *r->target_result =
         grpc_channel_args_copy_and_add(r->channel_args, &arg, 1);
-    grpc_exec_ctx_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE);
     r->next_completion = NULL;
   }
 }
diff --git a/test/core/end2end/fixtures/http_proxy.c b/test/core/end2end/fixtures/http_proxy.c
index 80865fc7a6..ca7d9e9f9a 100644
--- a/test/core/end2end/fixtures/http_proxy.c
+++ b/test/core/end2end/fixtures/http_proxy.c
@@ -376,15 +376,20 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg,
   gpr_ref_init(&conn->refcount, 1);
   conn->pollset_set = grpc_pollset_set_create();
   grpc_pollset_set_add_pollset(exec_ctx, conn->pollset_set, proxy->pollset);
-  grpc_closure_init(&conn->on_read_request_done, on_read_request_done, conn);
-  grpc_closure_init(&conn->on_server_connect_done, on_server_connect_done,
-                    conn);
-  grpc_closure_init(&conn->on_write_response_done, on_write_response_done,
-                    conn);
-  grpc_closure_init(&conn->on_client_read_done, on_client_read_done, conn);
-  grpc_closure_init(&conn->on_client_write_done, on_client_write_done, conn);
-  grpc_closure_init(&conn->on_server_read_done, on_server_read_done, conn);
-  grpc_closure_init(&conn->on_server_write_done, on_server_write_done, conn);
+  grpc_closure_init(&conn->on_read_request_done, on_read_request_done, conn,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&conn->on_server_connect_done, on_server_connect_done, conn,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&conn->on_write_response_done, on_write_response_done, conn,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&conn->on_client_read_done, on_client_read_done, conn,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&conn->on_client_write_done, on_client_write_done, conn,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&conn->on_server_read_done, on_server_read_done, conn,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&conn->on_server_write_done, on_server_write_done, conn,
+                    grpc_schedule_on_exec_ctx);
   grpc_slice_buffer_init(&conn->client_read_buffer);
   grpc_slice_buffer_init(&conn->client_deferred_write_buffer);
   grpc_slice_buffer_init(&conn->client_write_buffer);
@@ -471,7 +476,8 @@ void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy) {
   gpr_free(proxy->proxy_name);
   grpc_channel_args_destroy(proxy->channel_args);
   grpc_closure destroyed;
-  grpc_closure_init(&destroyed, destroy_pollset, proxy->pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, proxy->pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, proxy->pollset, &destroyed);
   gpr_free(proxy);
   grpc_exec_ctx_finish(&exec_ctx);
diff --git a/test/core/end2end/fuzzers/api_fuzzer.c b/test/core/end2end/fuzzers/api_fuzzer.c
index 746134c85b..8136f9312c 100644
--- a/test/core/end2end/fuzzers/api_fuzzer.c
+++ b/test/core/end2end/fuzzers/api_fuzzer.c
@@ -349,11 +349,11 @@ static void finish_resolve(grpc_exec_ctx *exec_ctx, void *arg,
     addrs->addrs = gpr_malloc(sizeof(*addrs->addrs));
     addrs->addrs[0].len = 0;
     *r->addrs = addrs;
-    grpc_exec_ctx_sched(exec_ctx, r->on_done, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, r->on_done, GRPC_ERROR_NONE);
   } else {
-    grpc_exec_ctx_sched(
+    grpc_closure_sched(
         exec_ctx, r->on_done,
-        GRPC_ERROR_CREATE_REFERENCING("Resolution failed", &error, 1), NULL);
+        GRPC_ERROR_CREATE_REFERENCING("Resolution failed", &error, 1));
   }
 
   gpr_free(r->addr);
@@ -398,7 +398,7 @@ static void do_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
   future_connect *fc = arg;
   if (error != GRPC_ERROR_NONE) {
     *fc->ep = NULL;
-    grpc_exec_ctx_sched(exec_ctx, fc->closure, GRPC_ERROR_REF(error), NULL);
+    grpc_closure_sched(exec_ctx, fc->closure, GRPC_ERROR_REF(error));
   } else if (g_server != NULL) {
     grpc_endpoint *client;
     grpc_endpoint *server;
@@ -410,7 +410,7 @@ static void do_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
     grpc_server_setup_transport(exec_ctx, g_server, transport, NULL, NULL);
     grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL);
 
-    grpc_exec_ctx_sched(exec_ctx, fc->closure, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, fc->closure, GRPC_ERROR_NONE);
   } else {
     sched_connect(exec_ctx, fc->closure, fc->ep, fc->deadline);
   }
@@ -421,8 +421,8 @@ static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
                           grpc_endpoint **ep, gpr_timespec deadline) {
   if (gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) < 0) {
     *ep = NULL;
-    grpc_exec_ctx_sched(exec_ctx, closure,
-                        GRPC_ERROR_CREATE("Connect deadline exceeded"), NULL);
+    grpc_closure_sched(exec_ctx, closure,
+                       GRPC_ERROR_CREATE("Connect deadline exceeded"));
     return;
   }
 
diff --git a/test/core/end2end/invalid_call_argument_test.c b/test/core/end2end/invalid_call_argument_test.c
index 765b6ad1be..d974d2c8ff 100644
--- a/test/core/end2end/invalid_call_argument_test.c
+++ b/test/core/end2end/invalid_call_argument_test.c
@@ -573,6 +573,29 @@ static void test_recv_close_on_server_twice() {
   cleanup_test();
 }
 
+static void test_invalid_initial_metadata_reserved_key() {
+  gpr_log(GPR_INFO, "test_invalid_initial_metadata_reserved_key");
+
+  grpc_metadata metadata;
+  metadata.key = ":start_with_colon";
+  metadata.value = "value";
+  metadata.value_length = 6;
+
+  grpc_op *op;
+  prepare_test(1);
+  op = g_state.ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 1;
+  op->data.send_initial_metadata.metadata = &metadata;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  GPR_ASSERT(GRPC_CALL_ERROR_INVALID_METADATA ==
+             grpc_call_start_batch(g_state.call, g_state.ops,
+                                   (size_t)(op - g_state.ops), tag(1), NULL));
+  cleanup_test();
+}
+
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   grpc_init();
@@ -595,6 +618,7 @@ int main(int argc, char **argv) {
   test_send_server_status_twice();
   test_recv_close_on_server_with_invalid_flags();
   test_recv_close_on_server_twice();
+  test_invalid_initial_metadata_reserved_key();
   grpc_shutdown();
 
   return 0;
diff --git a/test/core/end2end/tests/filter_causes_close.c b/test/core/end2end/tests/filter_causes_close.c
index 21905b98fa..7a7129ceb1 100644
--- a/test/core/end2end/tests/filter_causes_close.c
+++ b/test/core/end2end/tests/filter_causes_close.c
@@ -217,9 +217,9 @@ static void recv_im_ready(grpc_exec_ctx *exec_ctx, void *arg,
                                        &message);
     grpc_call_next_op(exec_ctx, elem, op);
   }
-  grpc_exec_ctx_sched(
+  grpc_closure_sched(
       exec_ctx, calld->recv_im_ready,
-      GRPC_ERROR_CREATE_REFERENCING("Forced call to close", &error, 1), NULL);
+      GRPC_ERROR_CREATE_REFERENCING("Forced call to close", &error, 1));
 }
 
 static void start_transport_stream_op(grpc_exec_ctx *exec_ctx,
@@ -228,7 +228,8 @@ static void start_transport_stream_op(grpc_exec_ctx *exec_ctx,
   call_data *calld = elem->call_data;
   if (op->recv_initial_metadata != NULL) {
     calld->recv_im_ready = op->recv_initial_metadata_ready;
-    op->recv_initial_metadata_ready = grpc_closure_create(recv_im_ready, elem);
+    op->recv_initial_metadata_ready =
+        grpc_closure_create(recv_im_ready, elem, grpc_schedule_on_exec_ctx);
   }
   grpc_call_next_op(exec_ctx, elem, op);
 }
diff --git a/test/core/http/httpcli_test.c b/test/core/http/httpcli_test.c
index 3e312c1dde..4f00cad205 100644
--- a/test/core/http/httpcli_test.c
+++ b/test/core/http/httpcli_test.c
@@ -90,9 +90,10 @@ static void test_get(int port) {
   grpc_http_response response;
   memset(&response, 0, sizeof(response));
   grpc_resource_quota *resource_quota = grpc_resource_quota_create("test_get");
-  grpc_httpcli_get(&exec_ctx, &g_context, &g_pops, resource_quota, &req,
-                   n_seconds_time(15),
-                   grpc_closure_create(on_finish, &response), &response);
+  grpc_httpcli_get(
+      &exec_ctx, &g_context, &g_pops, resource_quota, &req, n_seconds_time(15),
+      grpc_closure_create(on_finish, &response, grpc_schedule_on_exec_ctx),
+      &response);
   grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   gpr_mu_lock(g_mu);
   while (!g_done) {
@@ -130,9 +131,11 @@ static void test_post(int port) {
   grpc_http_response response;
   memset(&response, 0, sizeof(response));
   grpc_resource_quota *resource_quota = grpc_resource_quota_create("test_post");
-  grpc_httpcli_post(&exec_ctx, &g_context, &g_pops, resource_quota, &req,
-                    "hello", 5, n_seconds_time(15),
-                    grpc_closure_create(on_finish, &response), &response);
+  grpc_httpcli_post(
+      &exec_ctx, &g_context, &g_pops, resource_quota, &req, "hello", 5,
+      n_seconds_time(15),
+      grpc_closure_create(on_finish, &response, grpc_schedule_on_exec_ctx),
+      &response);
   grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   gpr_mu_lock(g_mu);
   while (!g_done) {
@@ -207,7 +210,8 @@ int main(int argc, char **argv) {
   test_post(port);
 
   grpc_httpcli_context_destroy(&g_context);
-  grpc_closure_init(&destroyed, destroy_pops, &g_pops);
+  grpc_closure_init(&destroyed, destroy_pops, &g_pops,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&g_pops),
                         &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
diff --git a/test/core/http/httpscli_test.c b/test/core/http/httpscli_test.c
index d06035149e..53b26b645f 100644
--- a/test/core/http/httpscli_test.c
+++ b/test/core/http/httpscli_test.c
@@ -91,9 +91,10 @@ static void test_get(int port) {
   grpc_http_response response;
   memset(&response, 0, sizeof(response));
   grpc_resource_quota *resource_quota = grpc_resource_quota_create("test_get");
-  grpc_httpcli_get(&exec_ctx, &g_context, &g_pops, resource_quota, &req,
-                   n_seconds_time(15),
-                   grpc_closure_create(on_finish, &response), &response);
+  grpc_httpcli_get(
+      &exec_ctx, &g_context, &g_pops, resource_quota, &req, n_seconds_time(15),
+      grpc_closure_create(on_finish, &response, grpc_schedule_on_exec_ctx),
+      &response);
   grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   gpr_mu_lock(g_mu);
   while (!g_done) {
@@ -132,9 +133,11 @@ static void test_post(int port) {
   grpc_http_response response;
   memset(&response, 0, sizeof(response));
   grpc_resource_quota *resource_quota = grpc_resource_quota_create("test_post");
-  grpc_httpcli_post(&exec_ctx, &g_context, &g_pops, resource_quota, &req,
-                    "hello", 5, n_seconds_time(15),
-                    grpc_closure_create(on_finish, &response), &response);
+  grpc_httpcli_post(
+      &exec_ctx, &g_context, &g_pops, resource_quota, &req, "hello", 5,
+      n_seconds_time(15),
+      grpc_closure_create(on_finish, &response, grpc_schedule_on_exec_ctx),
+      &response);
   grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   gpr_mu_lock(g_mu);
   while (!g_done) {
@@ -210,7 +213,8 @@ int main(int argc, char **argv) {
   test_post(port);
 
   grpc_httpcli_context_destroy(&g_context);
-  grpc_closure_init(&destroyed, destroy_pops, &g_pops);
+  grpc_closure_init(&destroyed, destroy_pops, &g_pops,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&g_pops),
                         &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
diff --git a/test/core/internal_api_canaries/iomgr.c b/test/core/internal_api_canaries/iomgr.c
index de03c47c13..773ef602b2 100644
--- a/test/core/internal_api_canaries/iomgr.c
+++ b/test/core/internal_api_canaries/iomgr.c
@@ -60,9 +60,9 @@ static void test_code(void) {
   closure_list.head = NULL;
   closure_list.tail = NULL;
 
-  grpc_closure_init(&closure, NULL, NULL);
+  grpc_closure_init(&closure, NULL, NULL, grpc_schedule_on_exec_ctx);
 
-  grpc_closure_create(NULL, NULL);
+  grpc_closure_create(NULL, NULL, grpc_schedule_on_exec_ctx);
 
   grpc_closure_list_move(NULL, NULL);
   grpc_closure_list_append(NULL, NULL, GRPC_ERROR_CREATE("Foo"));
@@ -72,8 +72,8 @@ static void test_code(void) {
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   grpc_exec_ctx_flush(&exec_ctx);
   grpc_exec_ctx_finish(&exec_ctx);
-  grpc_exec_ctx_sched(&exec_ctx, &closure, GRPC_ERROR_CREATE("Foo"), NULL);
-  grpc_exec_ctx_enqueue_list(&exec_ctx, &closure_list, NULL);
+  grpc_closure_sched(&exec_ctx, &closure, GRPC_ERROR_CREATE("Foo"));
+  grpc_closure_list_sched(&exec_ctx, &closure_list);
 
   /* endpoint.h */
   grpc_endpoint endpoint;
@@ -99,7 +99,6 @@ static void test_code(void) {
 
   /* executor.h */
   grpc_executor_init();
-  grpc_executor_push(&closure, GRPC_ERROR_CREATE("Phi"));
   grpc_executor_shutdown();
 
   /* pollset.h */
diff --git a/test/core/iomgr/combiner_test.c b/test/core/iomgr/combiner_test.c
index f7d5809be7..9b6d6ff9b4 100644
--- a/test/core/iomgr/combiner_test.c
+++ b/test/core/iomgr/combiner_test.c
@@ -59,9 +59,10 @@ static void test_execute_one(void) {
   grpc_combiner *lock = grpc_combiner_create(NULL);
   bool done = false;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  grpc_combiner_execute(&exec_ctx, lock,
-                        grpc_closure_create(set_bool_to_true, &done),
-                        GRPC_ERROR_NONE, false);
+  grpc_closure_sched(&exec_ctx,
+                     grpc_closure_create(set_bool_to_true, &done,
+                                         grpc_combiner_scheduler(lock, false)),
+                     GRPC_ERROR_NONE);
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(done);
   grpc_combiner_destroy(&exec_ctx, lock);
@@ -94,9 +95,10 @@ static void execute_many_loop(void *a) {
       ex_args *c = gpr_malloc(sizeof(*c));
       c->ctr = &args->ctr;
       c->value = n++;
-      grpc_combiner_execute(&exec_ctx, args->lock,
-                            grpc_closure_create(check_one, c), GRPC_ERROR_NONE,
-                            false);
+      grpc_closure_sched(
+          &exec_ctx, grpc_closure_create(check_one, c, grpc_combiner_scheduler(
+                                                           args->lock, false)),
+          GRPC_ERROR_NONE);
       grpc_exec_ctx_flush(&exec_ctx);
     }
     // sleep for a little bit, to test a combiner draining and another thread
@@ -134,9 +136,10 @@ static void in_finally(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
 }
 
 static void add_finally(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-  grpc_combiner_execute_finally(exec_ctx, arg,
-                                grpc_closure_create(in_finally, NULL),
-                                GRPC_ERROR_NONE, false);
+  grpc_closure_sched(exec_ctx, grpc_closure_create(
+                                   in_finally, NULL,
+                                   grpc_combiner_finally_scheduler(arg, false)),
+                     GRPC_ERROR_NONE);
 }
 
 static void test_execute_finally(void) {
@@ -144,8 +147,10 @@ static void test_execute_finally(void) {
 
   grpc_combiner *lock = grpc_combiner_create(NULL);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  grpc_combiner_execute(&exec_ctx, lock, grpc_closure_create(add_finally, lock),
-                        GRPC_ERROR_NONE, false);
+  grpc_closure_sched(&exec_ctx,
+                     grpc_closure_create(add_finally, lock,
+                                         grpc_combiner_scheduler(lock, false)),
+                     GRPC_ERROR_NONE);
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(got_in_finally);
   grpc_combiner_destroy(&exec_ctx, lock);
diff --git a/test/core/iomgr/endpoint_pair_test.c b/test/core/iomgr/endpoint_pair_test.c
index 2a257a7cea..f02171f2ef 100644
--- a/test/core/iomgr/endpoint_pair_test.c
+++ b/test/core/iomgr/endpoint_pair_test.c
@@ -81,7 +81,8 @@ int main(int argc, char **argv) {
   g_pollset = gpr_malloc(grpc_pollset_size());
   grpc_pollset_init(g_pollset, &g_mu);
   grpc_endpoint_tests(configs[0], g_pollset, g_mu);
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   grpc_shutdown();
diff --git a/test/core/iomgr/endpoint_tests.c b/test/core/iomgr/endpoint_tests.c
index 8186ea7e85..87a9d79e9b 100644
--- a/test/core/iomgr/endpoint_tests.c
+++ b/test/core/iomgr/endpoint_tests.c
@@ -211,9 +211,10 @@ static void read_and_write_test(grpc_endpoint_test_config config,
   state.write_done = 0;
   state.current_read_data = 0;
   state.current_write_data = 0;
-  grpc_closure_init(&state.done_read, read_and_write_test_read_handler, &state);
+  grpc_closure_init(&state.done_read, read_and_write_test_read_handler, &state,
+                    grpc_schedule_on_exec_ctx);
   grpc_closure_init(&state.done_write, read_and_write_test_write_handler,
-                    &state);
+                    &state, grpc_schedule_on_exec_ctx);
   grpc_slice_buffer_init(&state.outgoing);
   grpc_slice_buffer_init(&state.incoming);
 
@@ -290,16 +291,19 @@ static void multiple_shutdown_test(grpc_endpoint_test_config config) {
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   grpc_endpoint_add_to_pollset(&exec_ctx, f.client_ep, g_pollset);
   grpc_endpoint_read(&exec_ctx, f.client_ep, &slice_buffer,
-                     grpc_closure_create(inc_on_failure, &fail_count));
+                     grpc_closure_create(inc_on_failure, &fail_count,
+                                         grpc_schedule_on_exec_ctx));
   wait_for_fail_count(&exec_ctx, &fail_count, 0);
   grpc_endpoint_shutdown(&exec_ctx, f.client_ep);
   wait_for_fail_count(&exec_ctx, &fail_count, 1);
   grpc_endpoint_read(&exec_ctx, f.client_ep, &slice_buffer,
-                     grpc_closure_create(inc_on_failure, &fail_count));
+                     grpc_closure_create(inc_on_failure, &fail_count,
+                                         grpc_schedule_on_exec_ctx));
   wait_for_fail_count(&exec_ctx, &fail_count, 2);
   grpc_slice_buffer_add(&slice_buffer, grpc_slice_from_copied_string("a"));
   grpc_endpoint_write(&exec_ctx, f.client_ep, &slice_buffer,
-                      grpc_closure_create(inc_on_failure, &fail_count));
+                      grpc_closure_create(inc_on_failure, &fail_count,
+                                          grpc_schedule_on_exec_ctx));
   wait_for_fail_count(&exec_ctx, &fail_count, 3);
   grpc_endpoint_shutdown(&exec_ctx, f.client_ep);
   wait_for_fail_count(&exec_ctx, &fail_count, 3);
diff --git a/test/core/iomgr/ev_epoll_linux_test.c b/test/core/iomgr/ev_epoll_linux_test.c
index 564b05d7f4..5bce9801a5 100644
--- a/test/core/iomgr/ev_epoll_linux_test.c
+++ b/test/core/iomgr/ev_epoll_linux_test.c
@@ -102,7 +102,8 @@ static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx,
   int i;
 
   for (i = 0; i < num_pollsets; i++) {
-    grpc_closure_init(&destroyed, destroy_pollset, pollsets[i].pollset);
+    grpc_closure_init(&destroyed, destroy_pollset, pollsets[i].pollset,
+                      grpc_schedule_on_exec_ctx);
     grpc_pollset_shutdown(exec_ctx, pollsets[i].pollset, &destroyed);
 
     grpc_exec_ctx_flush(exec_ctx);
diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c
index 6166699fe6..4dd476526d 100644
--- a/test/core/iomgr/fd_posix_test.c
+++ b/test/core/iomgr/fd_posix_test.c
@@ -219,8 +219,8 @@ static void listen_cb(grpc_exec_ctx *exec_ctx, void *arg, /*=sv_arg*/
   se->sv = sv;
   se->em_fd = grpc_fd_create(fd, "listener");
   grpc_pollset_add_fd(exec_ctx, g_pollset, se->em_fd);
-  se->session_read_closure.cb = session_read_cb;
-  se->session_read_closure.cb_arg = se;
+  grpc_closure_init(&se->session_read_closure, session_read_cb, se,
+                    grpc_schedule_on_exec_ctx);
   grpc_fd_notify_on_read(exec_ctx, se->em_fd, &se->session_read_closure);
 
   grpc_fd_notify_on_read(exec_ctx, listen_em_fd, &sv->listen_closure);
@@ -249,8 +249,8 @@ static int server_start(grpc_exec_ctx *exec_ctx, server *sv) {
   sv->em_fd = grpc_fd_create(fd, "server");
   grpc_pollset_add_fd(exec_ctx, g_pollset, sv->em_fd);
   /* Register to be interested in reading from listen_fd. */
-  sv->listen_closure.cb = listen_cb;
-  sv->listen_closure.cb_arg = sv;
+  grpc_closure_init(&sv->listen_closure, listen_cb, sv,
+                    grpc_schedule_on_exec_ctx);
   grpc_fd_notify_on_read(exec_ctx, sv->em_fd, &sv->listen_closure);
 
   return port;
@@ -333,8 +333,8 @@ static void client_session_write(grpc_exec_ctx *exec_ctx, void *arg, /*client */
   if (errno == EAGAIN) {
     gpr_mu_lock(g_mu);
     if (cl->client_write_cnt < CLIENT_TOTAL_WRITE_CNT) {
-      cl->write_closure.cb = client_session_write;
-      cl->write_closure.cb_arg = cl;
+      grpc_closure_init(&cl->write_closure, client_session_write, cl,
+                        grpc_schedule_on_exec_ctx);
       grpc_fd_notify_on_write(exec_ctx, cl->em_fd, &cl->write_closure);
       cl->client_write_cnt++;
     } else {
@@ -459,10 +459,10 @@ static void test_grpc_fd_change(void) {
   grpc_closure second_closure;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
-  first_closure.cb = first_read_callback;
-  first_closure.cb_arg = &a;
-  second_closure.cb = second_read_callback;
-  second_closure.cb_arg = &b;
+  grpc_closure_init(&first_closure, first_read_callback, &a,
+                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&second_closure, second_read_callback, &b,
+                    grpc_schedule_on_exec_ctx);
 
   init_change_data(&a);
   init_change_data(&b);
@@ -546,7 +546,8 @@ int main(int argc, char **argv) {
   grpc_pollset_init(g_pollset, &g_mu);
   test_grpc_fd();
   test_grpc_fd_change();
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_free(g_pollset);
diff --git a/test/core/iomgr/resolve_address_test.c b/test/core/iomgr/resolve_address_test.c
index e4136a7a7a..d844e6eceb 100644
--- a/test/core/iomgr/resolve_address_test.c
+++ b/test/core/iomgr/resolve_address_test.c
@@ -71,7 +71,8 @@ void args_finish(grpc_exec_ctx *exec_ctx, args_struct *args) {
   grpc_pollset_set_del_pollset(exec_ctx, args->pollset_set, args->pollset);
   grpc_pollset_set_destroy(args->pollset_set);
   grpc_closure do_nothing_cb;
-  grpc_closure_init(&do_nothing_cb, do_nothing, NULL);
+  grpc_closure_init(&do_nothing_cb, do_nothing, NULL,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(exec_ctx, args->pollset, &do_nothing_cb);
   // exec_ctx needs to be flushed before calling grpc_pollset_destroy()
   grpc_exec_ctx_flush(exec_ctx);
@@ -136,8 +137,10 @@ static void test_localhost(void) {
   args_struct args;
   args_init(&exec_ctx, &args);
   poll_pollset_until_request_done(&args);
-  grpc_resolve_address(&exec_ctx, "localhost:1", NULL, args.pollset_set,
-                       grpc_closure_create(must_succeed, &args), &args.addrs);
+  grpc_resolve_address(
+      &exec_ctx, "localhost:1", NULL, args.pollset_set,
+      grpc_closure_create(must_succeed, &args, grpc_schedule_on_exec_ctx),
+      &args.addrs);
   args_finish(&exec_ctx, &args);
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -147,8 +150,10 @@ static void test_default_port(void) {
   args_struct args;
   args_init(&exec_ctx, &args);
   poll_pollset_until_request_done(&args);
-  grpc_resolve_address(&exec_ctx, "localhost", "1", args.pollset_set,
-                       grpc_closure_create(must_succeed, &args), &args.addrs);
+  grpc_resolve_address(
+      &exec_ctx, "localhost", "1", args.pollset_set,
+      grpc_closure_create(must_succeed, &args, grpc_schedule_on_exec_ctx),
+      &args.addrs);
   args_finish(&exec_ctx, &args);
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -158,8 +163,10 @@ static void test_missing_default_port(void) {
   args_struct args;
   args_init(&exec_ctx, &args);
   poll_pollset_until_request_done(&args);
-  grpc_resolve_address(&exec_ctx, "localhost", NULL, args.pollset_set,
-                       grpc_closure_create(must_fail, &args), &args.addrs);
+  grpc_resolve_address(
+      &exec_ctx, "localhost", NULL, args.pollset_set,
+      grpc_closure_create(must_fail, &args, grpc_schedule_on_exec_ctx),
+      &args.addrs);
   args_finish(&exec_ctx, &args);
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -169,8 +176,10 @@ static void test_ipv6_with_port(void) {
   args_struct args;
   args_init(&exec_ctx, &args);
   poll_pollset_until_request_done(&args);
-  grpc_resolve_address(&exec_ctx, "[2001:db8::1]:1", NULL, args.pollset_set,
-                       grpc_closure_create(must_succeed, &args), &args.addrs);
+  grpc_resolve_address(
+      &exec_ctx, "[2001:db8::1]:1", NULL, args.pollset_set,
+      grpc_closure_create(must_succeed, &args, grpc_schedule_on_exec_ctx),
+      &args.addrs);
   args_finish(&exec_ctx, &args);
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -185,8 +194,10 @@ static void test_ipv6_without_port(void) {
     args_struct args;
     args_init(&exec_ctx, &args);
     poll_pollset_until_request_done(&args);
-    grpc_resolve_address(&exec_ctx, kCases[i], "80", args.pollset_set,
-                         grpc_closure_create(must_succeed, &args), &args.addrs);
+    grpc_resolve_address(
+        &exec_ctx, kCases[i], "80", args.pollset_set,
+        grpc_closure_create(must_succeed, &args, grpc_schedule_on_exec_ctx),
+        &args.addrs);
     args_finish(&exec_ctx, &args);
     grpc_exec_ctx_finish(&exec_ctx);
   }
@@ -202,8 +213,10 @@ static void test_invalid_ip_addresses(void) {
     args_struct args;
     args_init(&exec_ctx, &args);
     poll_pollset_until_request_done(&args);
-    grpc_resolve_address(&exec_ctx, kCases[i], NULL, args.pollset_set,
-                         grpc_closure_create(must_fail, &args), &args.addrs);
+    grpc_resolve_address(
+        &exec_ctx, kCases[i], NULL, args.pollset_set,
+        grpc_closure_create(must_fail, &args, grpc_schedule_on_exec_ctx),
+        &args.addrs);
     args_finish(&exec_ctx, &args);
     grpc_exec_ctx_finish(&exec_ctx);
   }
@@ -219,8 +232,10 @@ static void test_unparseable_hostports(void) {
     args_struct args;
     args_init(&exec_ctx, &args);
     poll_pollset_until_request_done(&args);
-    grpc_resolve_address(&exec_ctx, kCases[i], "1", args.pollset_set,
-                         grpc_closure_create(must_fail, &args), &args.addrs);
+    grpc_resolve_address(
+        &exec_ctx, kCases[i], "1", args.pollset_set,
+        grpc_closure_create(must_fail, &args, grpc_schedule_on_exec_ctx),
+        &args.addrs);
     args_finish(&exec_ctx, &args);
     grpc_exec_ctx_finish(&exec_ctx);
   }
diff --git a/test/core/iomgr/resource_quota_test.c b/test/core/iomgr/resource_quota_test.c
index a82d44f7f8..181776341f 100644
--- a/test/core/iomgr/resource_quota_test.c
+++ b/test/core/iomgr/resource_quota_test.c
@@ -45,7 +45,9 @@ static void inc_int_cb(grpc_exec_ctx *exec_ctx, void *a, grpc_error *error) {
 static void set_bool_cb(grpc_exec_ctx *exec_ctx, void *a, grpc_error *error) {
   *(bool *)a = true;
 }
-grpc_closure *set_bool(bool *p) { return grpc_closure_create(set_bool_cb, p); }
+grpc_closure *set_bool(bool *p) {
+  return grpc_closure_create(set_bool_cb, p, grpc_schedule_on_exec_ctx);
+}
 
 typedef struct {
   size_t size;
@@ -67,7 +69,7 @@ grpc_closure *make_reclaimer(grpc_resource_user *resource_user, size_t size,
   a->size = size;
   a->resource_user = resource_user;
   a->then = then;
-  return grpc_closure_create(reclaimer_cb, a);
+  return grpc_closure_create(reclaimer_cb, a, grpc_schedule_on_exec_ctx);
 }
 
 static void unused_reclaimer_cb(grpc_exec_ctx *exec_ctx, void *arg,
@@ -76,7 +78,8 @@ static void unused_reclaimer_cb(grpc_exec_ctx *exec_ctx, void *arg,
   grpc_closure_run(exec_ctx, arg, GRPC_ERROR_NONE);
 }
 grpc_closure *make_unused_reclaimer(grpc_closure *then) {
-  return grpc_closure_create(unused_reclaimer_cb, then);
+  return grpc_closure_create(unused_reclaimer_cb, then,
+                             grpc_schedule_on_exec_ctx);
 }
 
 static void destroy_user(grpc_resource_user *usr) {
diff --git a/test/core/iomgr/tcp_client_posix_test.c b/test/core/iomgr/tcp_client_posix_test.c
index 5fab826fb7..0ea7a000eb 100644
--- a/test/core/iomgr/tcp_client_posix_test.c
+++ b/test/core/iomgr/tcp_client_posix_test.c
@@ -113,7 +113,7 @@ void test_succeeds(void) {
   /* connect to it */
   GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)addr,
                          (socklen_t *)&resolved_addr.len) == 0);
-  grpc_closure_init(&done, must_succeed, NULL);
+  grpc_closure_init(&done, must_succeed, NULL, grpc_schedule_on_exec_ctx);
   grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, g_pollset_set, NULL,
                           &resolved_addr, gpr_inf_future(GPR_CLOCK_REALTIME));
 
@@ -163,7 +163,7 @@ void test_fails(void) {
   gpr_mu_unlock(g_mu);
 
   /* connect to a broken address */
-  grpc_closure_init(&done, must_fail, NULL);
+  grpc_closure_init(&done, must_fail, NULL, grpc_schedule_on_exec_ctx);
   grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, g_pollset_set, NULL,
                           &resolved_addr, gpr_inf_future(GPR_CLOCK_REALTIME));
 
@@ -207,7 +207,8 @@ int main(int argc, char **argv) {
   gpr_log(GPR_ERROR, "End of first test");
   test_fails();
   grpc_pollset_set_destroy(g_pollset_set);
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   grpc_shutdown();
diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c
index 5eafa570bb..c646e61de0 100644
--- a/test/core/iomgr/tcp_posix_test.c
+++ b/test/core/iomgr/tcp_posix_test.c
@@ -194,7 +194,7 @@ static void read_test(size_t num_bytes, size_t slice_size) {
   state.read_bytes = 0;
   state.target_read_bytes = written_bytes;
   grpc_slice_buffer_init(&state.incoming);
-  grpc_closure_init(&state.read_cb, read_cb, &state);
+  grpc_closure_init(&state.read_cb, read_cb, &state, grpc_schedule_on_exec_ctx);
 
   grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);
 
@@ -245,7 +245,7 @@ static void large_read_test(size_t slice_size) {
   state.read_bytes = 0;
   state.target_read_bytes = (size_t)written_bytes;
   grpc_slice_buffer_init(&state.incoming);
-  grpc_closure_init(&state.read_cb, read_cb, &state);
+  grpc_closure_init(&state.read_cb, read_cb, &state, grpc_schedule_on_exec_ctx);
 
   grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);
 
@@ -384,7 +384,8 @@ static void write_test(size_t num_bytes, size_t slice_size) {
 
   grpc_slice_buffer_init(&outgoing);
   grpc_slice_buffer_addn(&outgoing, slices, num_blocks);
-  grpc_closure_init(&write_done_closure, write_done, &state);
+  grpc_closure_init(&write_done_closure, write_done, &state,
+                    grpc_schedule_on_exec_ctx);
 
   grpc_endpoint_write(&exec_ctx, ep, &outgoing, &write_done_closure);
   drain_socket_blocking(sv[0], num_bytes, num_bytes);
@@ -429,7 +430,8 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) {
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   grpc_closure fd_released_cb;
   int fd_released_done = 0;
-  grpc_closure_init(&fd_released_cb, &on_fd_released, &fd_released_done);
+  grpc_closure_init(&fd_released_cb, &on_fd_released, &fd_released_done,
+                    grpc_schedule_on_exec_ctx);
 
   gpr_log(GPR_INFO,
           "Release fd read_test of size %" PRIuPTR ", slice size %" PRIuPTR,
@@ -452,7 +454,7 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) {
   state.read_bytes = 0;
   state.target_read_bytes = written_bytes;
   grpc_slice_buffer_init(&state.incoming);
-  grpc_closure_init(&state.read_cb, read_cb, &state);
+  grpc_closure_init(&state.read_cb, read_cb, &state, grpc_schedule_on_exec_ctx);
 
   grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);
 
@@ -561,7 +563,8 @@ int main(int argc, char **argv) {
   grpc_pollset_init(g_pollset, &g_mu);
   grpc_endpoint_tests(configs[0], g_pollset, g_mu);
   run_tests();
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   grpc_shutdown();
diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c
index 9a7810e227..020f005980 100644
--- a/test/core/iomgr/tcp_server_posix_test.c
+++ b/test/core/iomgr/tcp_server_posix_test.c
@@ -104,7 +104,7 @@ static void server_weak_ref_shutdown(grpc_exec_ctx *exec_ctx, void *arg,
 static void server_weak_ref_init(server_weak_ref *weak_ref) {
   weak_ref->server = NULL;
   grpc_closure_init(&weak_ref->server_shutdown, server_weak_ref_shutdown,
-                    weak_ref);
+                    weak_ref, grpc_schedule_on_exec_ctx);
 }
 
 /* Make weak_ref->server_shutdown a shutdown_starting cb on server.
@@ -366,7 +366,8 @@ int main(int argc, char **argv) {
   test_connect(1);
   test_connect(10);
 
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   grpc_shutdown();
diff --git a/test/core/iomgr/udp_server_test.c b/test/core/iomgr/udp_server_test.c
index 9bea229466..0a247caf89 100644
--- a/test/core/iomgr/udp_server_test.c
+++ b/test/core/iomgr/udp_server_test.c
@@ -234,7 +234,8 @@ int main(int argc, char **argv) {
   test_receive(1);
   test_receive(10);
 
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_free(g_pollset);
diff --git a/test/core/security/create_jwt.c b/test/core/security/create_jwt.c
index 741ace9bdd..ac795f29d2 100644
--- a/test/core/security/create_jwt.c
+++ b/test/core/security/create_jwt.c
@@ -72,6 +72,7 @@ int main(int argc, char **argv) {
   char *scope = NULL;
   char *json_key_file_path = NULL;
   char *service_url = NULL;
+  grpc_init();
   gpr_cmdline *cl = gpr_cmdline_create("create_jwt");
   gpr_cmdline_add_string(cl, "json_key", "File path of the json key.",
                          &json_key_file_path);
@@ -102,5 +103,6 @@ int main(int argc, char **argv) {
   create_jwt(json_key_file_path, service_url, scope);
 
   gpr_cmdline_destroy(cl);
+  grpc_shutdown();
   return 0;
 }
diff --git a/test/core/security/credentials_test.c b/test/core/security/credentials_test.c
index d4c755088d..d624a38438 100644
--- a/test/core/security/credentials_test.c
+++ b/test/core/security/credentials_test.c
@@ -565,7 +565,7 @@ static int compute_engine_httpcli_get_success_override(
     grpc_httpcli_response *response) {
   validate_compute_engine_http_request(request);
   *response = http_response(200, valid_oauth2_json_response);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -575,7 +575,7 @@ static int compute_engine_httpcli_get_failure_override(
     grpc_httpcli_response *response) {
   validate_compute_engine_http_request(request);
   *response = http_response(403, "Not Authorized.");
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -668,7 +668,7 @@ static int refresh_token_httpcli_post_success(
     grpc_closure *on_done, grpc_httpcli_response *response) {
   validate_refresh_token_http_request(request, body, body_size);
   *response = http_response(200, valid_oauth2_json_response);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -678,7 +678,7 @@ static int refresh_token_httpcli_post_failure(
     grpc_closure *on_done, grpc_httpcli_response *response) {
   validate_refresh_token_http_request(request, body, body_size);
   *response = http_response(403, "Not Authorized.");
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -917,7 +917,7 @@ static int default_creds_gce_detection_httpcli_get_success_override(
   response->hdrs = headers;
   GPR_ASSERT(strcmp(request->http.path, "/") == 0);
   GPR_ASSERT(strcmp(request->host, "metadata.google.internal") == 0);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -975,7 +975,7 @@ static int default_creds_gce_detection_httpcli_get_failure_override(
   GPR_ASSERT(strcmp(request->http.path, "/") == 0);
   GPR_ASSERT(strcmp(request->host, "metadata.google.internal") == 0);
   *response = http_response(200, "");
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
diff --git a/test/core/security/jwt_verifier_test.c b/test/core/security/jwt_verifier_test.c
index f8afba8d6d..a4d65dccd9 100644
--- a/test/core/security/jwt_verifier_test.c
+++ b/test/core/security/jwt_verifier_test.c
@@ -166,6 +166,13 @@ static const char claims_without_time_constraint[] =
     "  \"jti\": \"jwtuniqueid\","
     "  \"foo\": \"bar\"}";
 
+static const char claims_with_bad_subject[] =
+    "{ \"aud\": \"https://foo.com\","
+    "  \"iss\": \"evil@blah.foo.com\","
+    "  \"sub\": \"juju@blah.foo.com\","
+    "  \"jti\": \"jwtuniqueid\","
+    "  \"foo\": \"bar\"}";
+
 static const char invalid_claims[] =
     "{ \"aud\": \"https://foo.com\","
     "  \"iss\": 46," /* Issuer cannot be a number. */
@@ -179,6 +186,38 @@ typedef struct {
   const char *expected_subject;
 } verifier_test_config;
 
+static void test_jwt_issuer_email_domain(void) {
+  const char *d = grpc_jwt_issuer_email_domain("https://foo.com");
+  GPR_ASSERT(d == NULL);
+  d = grpc_jwt_issuer_email_domain("foo.com");
+  GPR_ASSERT(d == NULL);
+  d = grpc_jwt_issuer_email_domain("");
+  GPR_ASSERT(d == NULL);
+  d = grpc_jwt_issuer_email_domain("@");
+  GPR_ASSERT(d == NULL);
+  d = grpc_jwt_issuer_email_domain("bar@foo");
+  GPR_ASSERT(strcmp(d, "foo") == 0);
+  d = grpc_jwt_issuer_email_domain("bar@foo.com");
+  GPR_ASSERT(strcmp(d, "foo.com") == 0);
+  d = grpc_jwt_issuer_email_domain("bar@blah.foo.com");
+  GPR_ASSERT(strcmp(d, "foo.com") == 0);
+  d = grpc_jwt_issuer_email_domain("bar.blah@blah.foo.com");
+  GPR_ASSERT(strcmp(d, "foo.com") == 0);
+  d = grpc_jwt_issuer_email_domain("bar.blah@baz.blah.foo.com");
+  GPR_ASSERT(strcmp(d, "foo.com") == 0);
+
+  /* This is not a very good parser but make sure we do not crash on these weird
+     inputs. */
+  d = grpc_jwt_issuer_email_domain("@foo");
+  GPR_ASSERT(strcmp(d, "foo") == 0);
+  d = grpc_jwt_issuer_email_domain("bar@.");
+  GPR_ASSERT(d != NULL);
+  d = grpc_jwt_issuer_email_domain("bar@..");
+  GPR_ASSERT(d != NULL);
+  d = grpc_jwt_issuer_email_domain("bar@...");
+  GPR_ASSERT(d != NULL);
+}
+
 static void test_claims_success(void) {
   grpc_jwt_claims *claims;
   grpc_slice s = grpc_slice_from_copied_string(claims_without_time_constraint);
@@ -242,6 +281,19 @@ static void test_bad_audience_claims_failure(void) {
   grpc_jwt_claims_destroy(claims);
 }
 
+static void test_bad_subject_claims_failure(void) {
+  grpc_jwt_claims *claims;
+  grpc_slice s = grpc_slice_from_copied_string(claims_with_bad_subject);
+  grpc_json *json = grpc_json_parse_string_with_len(
+      (char *)GRPC_SLICE_START_PTR(s), GRPC_SLICE_LENGTH(s));
+  GPR_ASSERT(json != NULL);
+  claims = grpc_jwt_claims_from_json(json, s);
+  GPR_ASSERT(claims != NULL);
+  GPR_ASSERT(grpc_jwt_claims_check(claims, "https://foo.com") ==
+             GRPC_JWT_VERIFIER_BAD_SUBJECT);
+  grpc_jwt_claims_destroy(claims);
+}
+
 static char *json_key_str(const char *last_part) {
   size_t result_len = strlen(json_key_str_part1) + strlen(json_key_str_part2) +
                       strlen(last_part);
@@ -294,7 +346,7 @@ static int httpcli_get_google_keys_for_email(
                     "/robot/v1/metadata/x509/"
                     "777-abaslkan11hlb6nmim3bpspl31ud@developer."
                     "gserviceaccount.com") == 0);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -338,7 +390,7 @@ static int httpcli_get_custom_keys_for_email(
   GPR_ASSERT(request->handshaker == &grpc_httpcli_ssl);
   GPR_ASSERT(strcmp(request->host, "keys.bar.com") == 0);
   GPR_ASSERT(strcmp(request->http.path, "/jwk/foo@bar.com") == 0);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -372,7 +424,7 @@ static int httpcli_get_jwk_set(grpc_exec_ctx *exec_ctx,
   GPR_ASSERT(request->handshaker == &grpc_httpcli_ssl);
   GPR_ASSERT(strcmp(request->host, "www.googleapis.com") == 0);
   GPR_ASSERT(strcmp(request->http.path, "/oauth2/v3/certs") == 0);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -387,7 +439,7 @@ static int httpcli_get_openid_config(grpc_exec_ctx *exec_ctx,
   GPR_ASSERT(strcmp(request->http.path, GRPC_OPENID_CONFIG_URL_SUFFIX) == 0);
   grpc_httpcli_set_override(httpcli_get_jwk_set,
                             httpcli_post_should_not_be_called);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -427,7 +479,7 @@ static int httpcli_get_bad_json(grpc_exec_ctx *exec_ctx,
                                 grpc_httpcli_response *response) {
   *response = http_response(200, gpr_strdup("{\"bad\": \"stuff\"}"));
   GPR_ASSERT(request->handshaker == &grpc_httpcli_ssl);
-  grpc_exec_ctx_sched(exec_ctx, on_done, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, on_done, GRPC_ERROR_NONE);
   return 1;
 }
 
@@ -563,10 +615,12 @@ static void test_jwt_verifier_bad_format(void) {
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   grpc_init();
+  test_jwt_issuer_email_domain();
   test_claims_success();
   test_expired_claims_failure();
   test_invalid_claims_failure();
   test_bad_audience_claims_failure();
+  test_bad_subject_claims_failure();
   test_jwt_verifier_google_email_issuer_success();
   test_jwt_verifier_custom_email_issuer_success();
   test_jwt_verifier_url_issuer_success();
diff --git a/test/core/security/oauth2_utils.c b/test/core/security/oauth2_utils.c
index 44a209258d..ff77af908a 100644
--- a/test/core/security/oauth2_utils.c
+++ b/test/core/security/oauth2_utils.c
@@ -92,7 +92,8 @@ char *grpc_test_fetch_oauth2_token_with_credentials(
   request.pops = grpc_polling_entity_create_from_pollset(pollset);
   request.is_done = 0;
 
-  grpc_closure_init(&do_nothing_closure, do_nothing, NULL);
+  grpc_closure_init(&do_nothing_closure, do_nothing, NULL,
+                    grpc_schedule_on_exec_ctx);
 
   grpc_call_credentials_get_request_metadata(
       &exec_ctx, creds, &request.pops, null_ctx, on_oauth2_response, &request);
diff --git a/test/core/security/secure_endpoint_test.c b/test/core/security/secure_endpoint_test.c
index b5d95004fe..cbf8a171af 100644
--- a/test/core/security/secure_endpoint_test.c
+++ b/test/core/security/secure_endpoint_test.c
@@ -158,7 +158,7 @@ static void test_leftover(grpc_endpoint_test_config config, size_t slice_size) {
   gpr_log(GPR_INFO, "Start test left over");
 
   grpc_slice_buffer_init(&incoming);
-  grpc_closure_init(&done_closure, inc_call_ctr, &n);
+  grpc_closure_init(&done_closure, inc_call_ctr, &n, grpc_schedule_on_exec_ctx);
   grpc_endpoint_read(&exec_ctx, f.client_ep, &incoming, &done_closure);
   grpc_exec_ctx_finish(&exec_ctx);
   GPR_ASSERT(n == 1);
@@ -191,7 +191,8 @@ int main(int argc, char **argv) {
   grpc_pollset_init(g_pollset, &g_mu);
   grpc_endpoint_tests(configs[0], g_pollset, g_mu);
   test_leftover(configs[1], 1);
-  grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
+  grpc_closure_init(&destroyed, destroy_pollset, g_pollset,
+                    grpc_schedule_on_exec_ctx);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);
   grpc_shutdown();
diff --git a/test/core/security/verify_jwt.c b/test/core/security/verify_jwt.c
index 043d29e6bb..ccc85c9f32 100644
--- a/test/core/security/verify_jwt.c
+++ b/test/core/security/verify_jwt.c
@@ -93,6 +93,7 @@ int main(int argc, char **argv) {
   char *aud = NULL;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
+  grpc_init();
   cl = gpr_cmdline_create("JWT verifier tool");
   gpr_cmdline_add_string(cl, "jwt", "JSON web token to verify", &jwt);
   gpr_cmdline_add_string(cl, "aud", "Audience for the JWT", &aud);
@@ -131,5 +132,6 @@ int main(int argc, char **argv) {
 
   grpc_jwt_verifier_destroy(verifier);
   gpr_cmdline_destroy(cl);
+  grpc_shutdown();
   return !sync.success;
 }
diff --git a/test/core/support/string_test.c b/test/core/support/string_test.c
index 78b77fad8e..af232db350 100644
--- a/test/core/support/string_test.c
+++ b/test/core/support/string_test.c
@@ -243,6 +243,8 @@ static void test_int64toa() {
 static void test_leftpad() {
   char *padded;
 
+  LOG_TEST_NAME("test_leftpad");
+
   padded = gpr_leftpad("foo", ' ', 5);
   GPR_ASSERT(0 == strcmp("  foo", padded));
   gpr_free(padded);
@@ -273,12 +275,25 @@ static void test_leftpad() {
 }
 
 static void test_stricmp(void) {
+  LOG_TEST_NAME("test_stricmp");
+
   GPR_ASSERT(0 == gpr_stricmp("hello", "hello"));
   GPR_ASSERT(0 == gpr_stricmp("HELLO", "hello"));
   GPR_ASSERT(gpr_stricmp("a", "b") < 0);
   GPR_ASSERT(gpr_stricmp("b", "a") > 0);
 }
 
+static void test_memrchr(void) {
+  LOG_TEST_NAME("test_memrchr");
+
+  GPR_ASSERT(NULL == gpr_memrchr(NULL, 'a', 0));
+  GPR_ASSERT(NULL == gpr_memrchr("", 'a', 0));
+  GPR_ASSERT(NULL == gpr_memrchr("hello", 'b', 5));
+  GPR_ASSERT(0 == strcmp((const char *)gpr_memrchr("hello", 'h', 5), "hello"));
+  GPR_ASSERT(0 == strcmp((const char *)gpr_memrchr("hello", 'o', 5), "o"));
+  GPR_ASSERT(0 == strcmp((const char *)gpr_memrchr("hello", 'l', 5), "lo"));
+}
+
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   test_strdup();
@@ -291,5 +306,6 @@ int main(int argc, char **argv) {
   test_int64toa();
   test_leftpad();
   test_stricmp();
+  test_memrchr();
   return 0;
 }
diff --git a/test/core/surface/concurrent_connectivity_test.c b/test/core/surface/concurrent_connectivity_test.c
index 93a4794222..8ebe8d07e4 100644
--- a/test/core/surface/concurrent_connectivity_test.c
+++ b/test/core/surface/concurrent_connectivity_test.c
@@ -229,9 +229,9 @@ int main(int argc, char **argv) {
   gpr_atm_rel_store(&args.stop, 1);
   gpr_thd_join(server);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  grpc_pollset_shutdown(
-      &exec_ctx, args.pollset,
-      grpc_closure_create(done_pollset_shutdown, args.pollset));
+  grpc_pollset_shutdown(&exec_ctx, args.pollset,
+                        grpc_closure_create(done_pollset_shutdown, args.pollset,
+                                            grpc_schedule_on_exec_ctx));
   grpc_exec_ctx_finish(&exec_ctx);
 
   grpc_shutdown();
diff --git a/test/core/surface/lame_client_test.c b/test/core/surface/lame_client_test.c
index 6afcefca92..b6db6a6b08 100644
--- a/test/core/surface/lame_client_test.c
+++ b/test/core/surface/lame_client_test.c
@@ -62,7 +62,8 @@ void test_transport_op(grpc_channel *channel) {
   grpc_connectivity_state state = GRPC_CHANNEL_IDLE;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
-  grpc_closure_init(&transport_op_cb, verify_connectivity, &state);
+  grpc_closure_init(&transport_op_cb, verify_connectivity, &state,
+                    grpc_schedule_on_exec_ctx);
 
   op = grpc_make_transport_op(NULL);
   op->on_connectivity_state_change = &transport_op_cb;
@@ -71,7 +72,8 @@ void test_transport_op(grpc_channel *channel) {
   elem->filter->start_transport_op(&exec_ctx, elem, op);
   grpc_exec_ctx_finish(&exec_ctx);
 
-  grpc_closure_init(&transport_op_cb, do_nothing, NULL);
+  grpc_closure_init(&transport_op_cb, do_nothing, NULL,
+                    grpc_schedule_on_exec_ctx);
   op = grpc_make_transport_op(&transport_op_cb);
   elem->filter->start_transport_op(&exec_ctx, elem, op);
   grpc_exec_ctx_finish(&exec_ctx);
diff --git a/test/core/transport/connectivity_state_test.c b/test/core/transport/connectivity_state_test.c
index 1050059eff..3520ef0a80 100644
--- a/test/core/transport/connectivity_state_test.c
+++ b/test/core/transport/connectivity_state_test.c
@@ -86,7 +86,8 @@ static void test_check(void) {
 
 static void test_subscribe_then_unsubscribe(void) {
   grpc_connectivity_state_tracker tracker;
-  grpc_closure *closure = grpc_closure_create(must_fail, THE_ARG);
+  grpc_closure *closure =
+      grpc_closure_create(must_fail, THE_ARG, grpc_schedule_on_exec_ctx);
   grpc_connectivity_state state = GRPC_CHANNEL_IDLE;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   gpr_log(GPR_DEBUG, "test_subscribe_then_unsubscribe");
@@ -109,7 +110,8 @@ static void test_subscribe_then_unsubscribe(void) {
 
 static void test_subscribe_then_destroy(void) {
   grpc_connectivity_state_tracker tracker;
-  grpc_closure *closure = grpc_closure_create(must_succeed, THE_ARG);
+  grpc_closure *closure =
+      grpc_closure_create(must_succeed, THE_ARG, grpc_schedule_on_exec_ctx);
   grpc_connectivity_state state = GRPC_CHANNEL_IDLE;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   gpr_log(GPR_DEBUG, "test_subscribe_then_destroy");
@@ -128,7 +130,8 @@ static void test_subscribe_then_destroy(void) {
 
 static void test_subscribe_with_failure_then_destroy(void) {
   grpc_connectivity_state_tracker tracker;
-  grpc_closure *closure = grpc_closure_create(must_fail, THE_ARG);
+  grpc_closure *closure =
+      grpc_closure_create(must_fail, THE_ARG, grpc_schedule_on_exec_ctx);
   grpc_connectivity_state state = GRPC_CHANNEL_SHUTDOWN;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   gpr_log(GPR_DEBUG, "test_subscribe_with_failure_then_destroy");
diff --git a/test/core/util/mock_endpoint.c b/test/core/util/mock_endpoint.c
index bf6d85252a..04793bceab 100644
--- a/test/core/util/mock_endpoint.c
+++ b/test/core/util/mock_endpoint.c
@@ -55,7 +55,7 @@ static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   gpr_mu_lock(&m->mu);
   if (m->read_buffer.count > 0) {
     grpc_slice_buffer_swap(&m->read_buffer, slices);
-    grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_NONE);
   } else {
     m->on_read = cb;
     m->on_read_out = slices;
@@ -69,7 +69,7 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   for (size_t i = 0; i < slices->count; i++) {
     m->on_write(slices->slices[i]);
   }
-  grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
+  grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_NONE);
 }
 
 static void me_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
@@ -82,8 +82,8 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   grpc_mock_endpoint *m = (grpc_mock_endpoint *)ep;
   gpr_mu_lock(&m->mu);
   if (m->on_read) {
-    grpc_exec_ctx_sched(exec_ctx, m->on_read,
-                        GRPC_ERROR_CREATE("Endpoint Shutdown"), NULL);
+    grpc_closure_sched(exec_ctx, m->on_read,
+                       GRPC_ERROR_CREATE("Endpoint Shutdown"));
     m->on_read = NULL;
   }
   gpr_mu_unlock(&m->mu);
@@ -144,7 +144,7 @@ void grpc_mock_endpoint_put_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   gpr_mu_lock(&m->mu);
   if (m->on_read != NULL) {
     grpc_slice_buffer_add(m->on_read_out, slice);
-    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE);
     m->on_read = NULL;
   } else {
     grpc_slice_buffer_add(&m->read_buffer, slice);
diff --git a/test/core/util/passthru_endpoint.c b/test/core/util/passthru_endpoint.c
index b3405f02e9..15ba092c5b 100644
--- a/test/core/util/passthru_endpoint.c
+++ b/test/core/util/passthru_endpoint.c
@@ -63,11 +63,10 @@ static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   half *m = (half *)ep;
   gpr_mu_lock(&m->parent->mu);
   if (m->parent->shutdown) {
-    grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_CREATE("Already shutdown"),
-                        NULL);
+    grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_CREATE("Already shutdown"));
   } else if (m->read_buffer.count > 0) {
     grpc_slice_buffer_swap(&m->read_buffer, slices);
-    grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_NONE);
   } else {
     m->on_read = cb;
     m->on_read_out = slices;
@@ -91,7 +90,7 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
     for (size_t i = 0; i < slices->count; i++) {
       grpc_slice_buffer_add(m->on_read_out, grpc_slice_ref(slices->slices[i]));
     }
-    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
+    grpc_closure_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE);
     m->on_read = NULL;
   } else {
     for (size_t i = 0; i < slices->count; i++) {
@@ -99,7 +98,7 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
     }
   }
   gpr_mu_unlock(&m->parent->mu);
-  grpc_exec_ctx_sched(exec_ctx, cb, error, NULL);
+  grpc_closure_sched(exec_ctx, cb, error);
 }
 
 static void me_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
@@ -113,14 +112,12 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   gpr_mu_lock(&m->parent->mu);
   m->parent->shutdown = true;
   if (m->on_read) {
-    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"),
-                        NULL);
+    grpc_closure_sched(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"));
     m->on_read = NULL;
   }
   m = other_half(m);
   if (m->on_read) {
-    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"),
-                        NULL);
+    grpc_closure_sched(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"));
     m->on_read = NULL;
   }
   gpr_mu_unlock(&m->parent->mu);
diff --git a/test/core/util/port_server_client.c b/test/core/util/port_server_client.c
index b2342feeb4..0bde726ba1 100644
--- a/test/core/util/port_server_client.c
+++ b/test/core/util/port_server_client.c
@@ -92,7 +92,8 @@ void grpc_free_port_using_server(char *server, int port) {
   grpc_pollset *pollset = gpr_malloc(grpc_pollset_size());
   grpc_pollset_init(pollset, &pr.mu);
   pr.pops = grpc_polling_entity_create_from_pollset(pollset);
-  shutdown_closure = grpc_closure_create(destroy_pops_and_shutdown, &pr.pops);
+  shutdown_closure = grpc_closure_create(destroy_pops_and_shutdown, &pr.pops,
+                                         grpc_schedule_on_exec_ctx);
 
   req.host = server;
   gpr_asprintf(&path, "/drop/%d", port);
@@ -103,7 +104,9 @@ void grpc_free_port_using_server(char *server, int port) {
       grpc_resource_quota_create("port_server_client/free");
   grpc_httpcli_get(&exec_ctx, &context, &pr.pops, resource_quota, &req,
                    GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
-                   grpc_closure_create(freed_port_from_server, &pr), &rsp);
+                   grpc_closure_create(freed_port_from_server, &pr,
+                                       grpc_schedule_on_exec_ctx),
+                   &rsp);
   grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   gpr_mu_lock(pr.mu);
   while (!pr.done) {
@@ -174,7 +177,8 @@ static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
         grpc_resource_quota_create("port_server_client/pick_retry");
     grpc_httpcli_get(exec_ctx, pr->ctx, &pr->pops, resource_quota, &req,
                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
-                     grpc_closure_create(got_port_from_server, pr),
+                     grpc_closure_create(got_port_from_server, pr,
+                                         grpc_schedule_on_exec_ctx),
                      &pr->response);
     grpc_resource_quota_internal_unref(exec_ctx, resource_quota);
     return;
@@ -208,7 +212,8 @@ int grpc_pick_port_using_server(char *server) {
   grpc_pollset *pollset = gpr_malloc(grpc_pollset_size());
   grpc_pollset_init(pollset, &pr.mu);
   pr.pops = grpc_polling_entity_create_from_pollset(pollset);
-  shutdown_closure = grpc_closure_create(destroy_pops_and_shutdown, &pr.pops);
+  shutdown_closure = grpc_closure_create(destroy_pops_and_shutdown, &pr.pops,
+                                         grpc_schedule_on_exec_ctx);
   pr.port = -1;
   pr.server = server;
   pr.ctx = &context;
@@ -219,10 +224,11 @@ int grpc_pick_port_using_server(char *server) {
   grpc_httpcli_context_init(&context);
   grpc_resource_quota *resource_quota =
       grpc_resource_quota_create("port_server_client/pick");
-  grpc_httpcli_get(&exec_ctx, &context, &pr.pops, resource_quota, &req,
-                   GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
-                   grpc_closure_create(got_port_from_server, &pr),
-                   &pr.response);
+  grpc_httpcli_get(
+      &exec_ctx, &context, &pr.pops, resource_quota, &req,
+      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
+      grpc_closure_create(got_port_from_server, &pr, grpc_schedule_on_exec_ctx),
+      &pr.response);
   grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_mu_lock(pr.mu);
diff --git a/test/core/util/test_tcp_server.c b/test/core/util/test_tcp_server.c
index 16df91d968..2338b81ab1 100644
--- a/test/core/util/test_tcp_server.c
+++ b/test/core/util/test_tcp_server.c
@@ -57,7 +57,8 @@ void test_tcp_server_init(test_tcp_server *server,
                           grpc_tcp_server_cb on_connect, void *user_data) {
   grpc_init();
   server->tcp_server = NULL;
-  grpc_closure_init(&server->shutdown_complete, on_server_destroyed, server);
+  grpc_closure_init(&server->shutdown_complete, on_server_destroyed, server,
+                    grpc_schedule_on_exec_ctx);
   server->shutdown = 0;
   server->pollset = gpr_malloc(grpc_pollset_size());
   grpc_pollset_init(server->pollset, &server->mu);
@@ -111,7 +112,8 @@ void test_tcp_server_destroy(test_tcp_server *server) {
   gpr_timespec shutdown_deadline;
   grpc_closure do_nothing_cb;
   grpc_tcp_server_unref(&exec_ctx, server->tcp_server);
-  grpc_closure_init(&do_nothing_cb, do_nothing, NULL);
+  grpc_closure_init(&do_nothing_cb, do_nothing, NULL,
+                    grpc_schedule_on_exec_ctx);
   shutdown_deadline = gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
                                    gpr_time_from_seconds(5, GPR_TIMESPAN));
   while (!server->shutdown &&
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index 22b2cd080d..93ef32db77 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -44,6 +44,7 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/host_port.h>
 #include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
 
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/support/env.h"
@@ -99,23 +100,36 @@ static std::unordered_map<string, std::deque<int>> get_hosts_and_cores(
   return hosts;
 }
 
-static deque<string> get_workers(const string& name) {
-  char* env = gpr_getenv(name.c_str());
-  if (!env || strlen(env) == 0) return deque<string>();
-
+static deque<string> get_workers(const string& env_name) {
+  char* env = gpr_getenv(env_name.c_str());
+  if (!env) {
+    env = gpr_strdup("");
+  }
   deque<string> out;
   char* p = env;
-  for (;;) {
-    char* comma = strchr(p, ',');
-    if (comma) {
-      out.emplace_back(p, comma);
-      p = comma + 1;
-    } else {
-      out.emplace_back(p);
-      gpr_free(env);
-      return out;
+  if (strlen(env) != 0) {
+    for (;;) {
+      char* comma = strchr(p, ',');
+      if (comma) {
+        out.emplace_back(p, comma);
+        p = comma + 1;
+      } else {
+        out.emplace_back(p);
+        break;
+      }
     }
   }
+  if (out.size() == 0) {
+    gpr_log(GPR_ERROR,
+            "Environment variable \"%s\" does not contain a list of QPS "
+            "workers to use. Set it to a comma-separated list of "
+            "hostname:port pairs, starting with hosts that should act as "
+            "servers. E.g. export "
+            "%s=\"serverhost1:1234,clienthost1:1234,clienthost2:1234\"",
+            env_name.c_str(), env_name.c_str());
+  }
+  gpr_free(env);
+  return out;
 }
 
 // helpers for postprocess_scenario_result
@@ -195,7 +209,8 @@ static void postprocess_scenario_result(ScenarioResult* result) {
 std::unique_ptr<ScenarioResult> RunScenario(
     const ClientConfig& initial_client_config, size_t num_clients,
     const ServerConfig& initial_server_config, size_t num_servers,
-    int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count) {
+    int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
+    const char* qps_server_target_override, bool configure_core_lists) {
   // Log everything from the driver
   gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
 
@@ -240,9 +255,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
       workers.push_back(addr);
     }
   }
-
-  // Setup the hosts and core counts
-  auto hosts_cores = get_hosts_and_cores(workers);
+  GPR_ASSERT(workers.size() != 0);
 
   // if num_clients is set to <=0, do dynamic sizing: all workers
   // except for servers are clients
@@ -264,6 +277,11 @@ std::unique_ptr<ScenarioResult> RunScenario(
     unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
   };
   std::vector<ServerData> servers(num_servers);
+  std::unordered_map<string, std::deque<int>> hosts_cores;
+
+  if (configure_core_lists) {
+    hosts_cores = get_hosts_and_cores(workers);
+  }
   for (size_t i = 0; i < num_servers; i++) {
     gpr_log(GPR_INFO, "Starting server on %s (worker #%" PRIuPTR ")",
             workers[i].c_str(), i);
@@ -271,37 +289,36 @@ std::unique_ptr<ScenarioResult> RunScenario(
         CreateChannel(workers[i], InsecureChannelCredentials()));
 
     ServerConfig server_config = initial_server_config;
-    char* host;
-    char* driver_port;
-    char* cli_target;
-    gpr_split_host_port(workers[i].c_str(), &host, &driver_port);
-    string host_str(host);
     int server_core_limit = initial_server_config.core_limit();
     int client_core_limit = initial_client_config.core_limit();
 
-    if (server_core_limit == 0 && client_core_limit > 0) {
-      // In this case, limit the server cores if it matches the
-      // same host as one or more clients
-      const auto& dq = hosts_cores.at(host_str);
-      bool match = false;
-      int limit = dq.size();
-      for (size_t cli = 0; cli < num_clients; cli++) {
-        if (host_str == get_host(workers[cli + num_servers])) {
-          limit -= client_core_limit;
-          match = true;
+    if (configure_core_lists) {
+      string host_str(get_host(workers[i]));
+      if (server_core_limit == 0 && client_core_limit > 0) {
+        // In this case, limit the server cores if it matches the
+        // same host as one or more clients
+        const auto& dq = hosts_cores.at(host_str);
+        bool match = false;
+        int limit = dq.size();
+        for (size_t cli = 0; cli < num_clients; cli++) {
+          if (host_str == get_host(workers[cli + num_servers])) {
+            limit -= client_core_limit;
+            match = true;
+          }
+        }
+        if (match) {
+          GPR_ASSERT(limit > 0);
+          server_core_limit = limit;
         }
       }
-      if (match) {
-        GPR_ASSERT(limit > 0);
-        server_core_limit = limit;
-      }
-    }
-    if (server_core_limit > 0) {
-      auto& dq = hosts_cores.at(host_str);
-      GPR_ASSERT(dq.size() >= static_cast<size_t>(server_core_limit));
-      for (int core = 0; core < server_core_limit; core++) {
-        server_config.add_core_list(dq.front());
-        dq.pop_front();
+      if (server_core_limit > 0) {
+        auto& dq = hosts_cores.at(host_str);
+        GPR_ASSERT(dq.size() >= static_cast<size_t>(server_core_limit));
+        gpr_log(GPR_INFO, "Setting server core_list");
+        for (int core = 0; core < server_core_limit; core++) {
+          server_config.add_core_list(dq.front());
+          dq.pop_front();
+        }
       }
     }
 
@@ -315,11 +332,19 @@ std::unique_ptr<ScenarioResult> RunScenario(
     if (!servers[i].stream->Read(&init_status)) {
       gpr_log(GPR_ERROR, "Server %zu did not yield initial status", i);
     }
-    gpr_join_host_port(&cli_target, host, init_status.port());
-    client_config.add_server_targets(cli_target);
-    gpr_free(host);
-    gpr_free(driver_port);
-    gpr_free(cli_target);
+    if (qps_server_target_override != NULL &&
+        strlen(qps_server_target_override) > 0) {
+      // overriding the qps server target only works if there is 1 server
+      GPR_ASSERT(num_servers == 1);
+      client_config.add_server_targets(qps_server_target_override);
+    } else {
+      std::string host;
+      char* cli_target;
+      host = get_host(workers[i]);
+      gpr_join_host_port(&cli_target, host.c_str(), init_status.port());
+      client_config.add_server_targets(cli_target);
+      gpr_free(cli_target);
+    }
   }
 
   // Targets are all set by now
@@ -341,7 +366,8 @@ std::unique_ptr<ScenarioResult> RunScenario(
 
     int server_core_limit = initial_server_config.core_limit();
     int client_core_limit = initial_client_config.core_limit();
-    if ((server_core_limit > 0) || (client_core_limit > 0)) {
+    if (configure_core_lists &&
+        ((server_core_limit > 0) || (client_core_limit > 0))) {
       auto& dq = hosts_cores.at(get_host(worker));
       if (client_core_limit == 0) {
         // limit client cores if it matches a server host
@@ -359,6 +385,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
       }
       if (client_core_limit > 0) {
         GPR_ASSERT(dq.size() >= static_cast<size_t>(client_core_limit));
+        gpr_log(GPR_INFO, "Setting client core_list");
         for (int core = 0; core < client_core_limit; core++) {
           per_client_config.add_core_list(dq.front());
           dq.pop_front();
@@ -548,6 +575,9 @@ bool RunQuit() {
   // Get client, server lists
   bool result = true;
   auto workers = get_workers("QPS_WORKERS");
+  if (workers.size() == 0) {
+    return false;
+  }
   for (size_t i = 0; i < workers.size(); i++) {
     auto stub = WorkerService::NewStub(
         CreateChannel(workers[i], InsecureChannelCredentials()));
diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h
index 93f4370caf..b5c8152e1b 100644
--- a/test/cpp/qps/driver.h
+++ b/test/cpp/qps/driver.h
@@ -45,7 +45,9 @@ namespace testing {
 std::unique_ptr<ScenarioResult> RunScenario(
     const grpc::testing::ClientConfig& client_config, size_t num_clients,
     const grpc::testing::ServerConfig& server_config, size_t num_servers,
-    int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count);
+    int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
+    const char* qps_server_target_override = "",
+    bool configure_core_lists = true);
 
 bool RunQuit();
 }  // namespace testing
diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc
index 31b5917fb7..da835b995a 100644
--- a/test/cpp/qps/qps_json_driver.cc
+++ b/test/cpp/qps/qps_json_driver.cc
@@ -67,17 +67,25 @@ DEFINE_double(error_tolerance, 0.01,
               "range is narrower than the error_tolerance computed range, we "
               "stop the search.");
 
+DEFINE_string(qps_server_target_override, "",
+              "Override QPS server target to configure in client configs."
+              "Only applicable if there is a single benchmark server.");
+DEFINE_bool(configure_core_lists, true,
+            "Provide 'core_list' parameters to workers. Value determined "
+            "by cores available and 'core_limit' parameters of the scenarios.");
+
 namespace grpc {
 namespace testing {
 
 static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario,
                                                     bool* success) {
   std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
-  auto result =
-      RunScenario(scenario.client_config(), scenario.num_clients(),
-                  scenario.server_config(), scenario.num_servers(),
-                  scenario.warmup_seconds(), scenario.benchmark_seconds(),
-                  scenario.spawn_local_worker_count());
+  auto result = RunScenario(
+      scenario.client_config(), scenario.num_clients(),
+      scenario.server_config(), scenario.num_servers(),
+      scenario.warmup_seconds(), scenario.benchmark_seconds(),
+      scenario.spawn_local_worker_count(),
+      FLAGS_qps_server_target_override.c_str(), FLAGS_configure_core_lists);
 
   // Amend the result with scenario config. Eventually we should adjust
   // RunScenario contract so we don't need to touch the result here.