Merge branch 'master' of https://github.com/grpc/grpc into channelz-subchannels

27 files changed, 1125 insertions, 206 deletions

diff --git a/test/core/bad_client/bad_client.cc b/test/core/bad_client/bad_client.cc
index c03ebcf409..ade23133c5 100644
--- a/test/core/bad_client/bad_client.cc
+++ b/test/core/bad_client/bad_client.cc
@@ -115,7 +115,7 @@ void grpc_run_client_side_validator(grpc_bad_client_arg* arg, uint32_t flags,
                     grpc_schedule_on_exec_ctx);
 
   /* Write data */
-  grpc_endpoint_write(sfd->client, &outgoing, &done_write_closure);
+  grpc_endpoint_write(sfd->client, &outgoing, &done_write_closure, nullptr);
   grpc_core::ExecCtx::Get()->Flush();
 
   /* Await completion, unless the request is large and write may not finish
diff --git a/test/core/channel/minimal_stack_is_minimal_test.cc b/test/core/channel/minimal_stack_is_minimal_test.cc
index 5b651ed39b..e5953acedc 100644
--- a/test/core/channel/minimal_stack_is_minimal_test.cc
+++ b/test/core/channel/minimal_stack_is_minimal_test.cc
@@ -85,21 +85,21 @@ int main(int argc, char** argv) {
 
   // tests with a default stack
   errors +=
-      CHECK_STACK("unknown", nullptr, GRPC_CLIENT_DIRECT_CHANNEL, "deadline",
-                  "authority", "message_size", "connected", NULL);
+      CHECK_STACK("unknown", nullptr, GRPC_CLIENT_DIRECT_CHANNEL, "authority",
+                  "message_size", "deadline", "connected", NULL);
   errors += CHECK_STACK("unknown", nullptr, GRPC_CLIENT_SUBCHANNEL, "authority",
                         "message_size", "connected", NULL);
   errors += CHECK_STACK("unknown", nullptr, GRPC_SERVER_CHANNEL, "server",
-                        "deadline", "message_size", "connected", NULL);
+                        "message_size", "deadline", "connected", NULL);
   errors += CHECK_STACK("chttp2", nullptr, GRPC_CLIENT_DIRECT_CHANNEL,
-                        "deadline", "authority", "message_size",
-                        "message_compress", "http-client", "connected", NULL);
+                        "authority", "message_size", "deadline", "http-client",
+                        "message_compress", "connected", NULL);
   errors += CHECK_STACK("chttp2", nullptr, GRPC_CLIENT_SUBCHANNEL, "authority",
-                        "message_size", "message_compress", "http-client",
+                        "message_size", "http-client", "message_compress",
                         "connected", NULL);
   errors += CHECK_STACK("chttp2", nullptr, GRPC_SERVER_CHANNEL, "server",
-                        "deadline", "message_size", "message_compress",
-                        "http-server", "connected", NULL);
+                        "message_size", "deadline", "http-server",
+                        "message_compress", "connected", NULL);
   errors += CHECK_STACK(nullptr, nullptr, GRPC_CLIENT_CHANNEL, "client-channel",
                         NULL);
 
diff --git a/test/core/client_channel/resolvers/dns_resolver_cooldown_test.cc b/test/core/client_channel/resolvers/dns_resolver_cooldown_test.cc
index b1f3a1c08a..1a7db40f59 100644
--- a/test/core/client_channel/resolvers/dns_resolver_cooldown_test.cc
+++ b/test/core/client_channel/resolvers/dns_resolver_cooldown_test.cc
@@ -28,12 +28,16 @@
 #include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "test/core/util/test_config.h"
 
+constexpr int kMinResolutionPeriodMs = 1000;
+// Provide some slack when checking intervals, to allow for test timing issues.
+constexpr int kMinResolutionPeriodForCheckMs = 900;
+
 extern grpc_address_resolver_vtable* grpc_resolve_address_impl;
 static grpc_address_resolver_vtable* default_resolve_address;
 
 static grpc_combiner* g_combiner;
 
-grpc_ares_request* (*g_default_dns_lookup_ares_locked)(
+static grpc_ares_request* (*g_default_dns_lookup_ares_locked)(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
@@ -43,7 +47,7 @@ grpc_ares_request* (*g_default_dns_lookup_ares_locked)(
 // times a system-level resolution has happened.
 static int g_resolution_count;
 
-struct iomgr_args {
+static struct iomgr_args {
   gpr_event ev;
   gpr_atm done_atm;
   gpr_mu* mu;
@@ -61,6 +65,16 @@ static void test_resolve_address_impl(const char* name,
   default_resolve_address->resolve_address(
       name, default_port, g_iomgr_args.pollset_set, on_done, addrs);
   ++g_resolution_count;
+  static grpc_millis last_resolution_time = 0;
+  if (last_resolution_time == 0) {
+    last_resolution_time =
+        grpc_timespec_to_millis_round_up(gpr_now(GPR_CLOCK_MONOTONIC));
+  } else {
+    grpc_millis now =
+        grpc_timespec_to_millis_round_up(gpr_now(GPR_CLOCK_MONOTONIC));
+    GPR_ASSERT(now - last_resolution_time >= kMinResolutionPeriodForCheckMs);
+    last_resolution_time = now;
+  }
 }
 
 static grpc_error* test_blocking_resolve_address_impl(
@@ -73,7 +87,7 @@ static grpc_error* test_blocking_resolve_address_impl(
 static grpc_address_resolver_vtable test_resolver = {
     test_resolve_address_impl, test_blocking_resolve_address_impl};
 
-grpc_ares_request* test_dns_lookup_ares_locked(
+static grpc_ares_request* test_dns_lookup_ares_locked(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
@@ -82,6 +96,16 @@ grpc_ares_request* test_dns_lookup_ares_locked(
       dns_server, name, default_port, g_iomgr_args.pollset_set, on_done, addrs,
       check_grpclb, service_config_json, combiner);
   ++g_resolution_count;
+  static grpc_millis last_resolution_time = 0;
+  if (last_resolution_time == 0) {
+    last_resolution_time =
+        grpc_timespec_to_millis_round_up(gpr_now(GPR_CLOCK_MONOTONIC));
+  } else {
+    grpc_millis now =
+        grpc_timespec_to_millis_round_up(gpr_now(GPR_CLOCK_MONOTONIC));
+    GPR_ASSERT(now - last_resolution_time >= kMinResolutionPeriodForCheckMs);
+    last_resolution_time = now;
+  }
   return result;
 }
 
@@ -91,7 +115,7 @@ static gpr_timespec test_deadline(void) {
 
 static void do_nothing(void* arg, grpc_error* error) {}
 
-void iomgr_args_init(iomgr_args* args) {
+static void iomgr_args_init(iomgr_args* args) {
   gpr_event_init(&args->ev);
   args->pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
   grpc_pollset_init(args->pollset, &args->mu);
@@ -100,7 +124,7 @@ void iomgr_args_init(iomgr_args* args) {
   gpr_atm_rel_store(&args->done_atm, 0);
 }
 
-void iomgr_args_finish(iomgr_args* args) {
+static void iomgr_args_finish(iomgr_args* args) {
   GPR_ASSERT(gpr_event_wait(&args->ev, test_deadline()));
   grpc_pollset_set_del_pollset(args->pollset_set, args->pollset);
   grpc_pollset_set_destroy(args->pollset_set);
@@ -146,29 +170,19 @@ struct OnResolutionCallbackArg {
   const char* uri_str = nullptr;
   grpc_core::OrphanablePtr<grpc_core::Resolver> resolver;
   grpc_channel_args* result = nullptr;
-  grpc_millis delay_before_second_resolution = 0;
 };
 
-// Counter for the number of times a resolution notification callback has been
-// invoked.
-static int g_on_resolution_invocations_count;
-
 // Set to true by the last callback in the resolution chain.
-bool g_all_callbacks_invoked;
+static bool g_all_callbacks_invoked;
 
-void on_fourth_resolution(void* arg, grpc_error* error) {
+static void on_second_resolution(void* arg, grpc_error* error) {
   OnResolutionCallbackArg* cb_arg = static_cast<OnResolutionCallbackArg*>(arg);
   grpc_channel_args_destroy(cb_arg->result);
   GPR_ASSERT(error == GRPC_ERROR_NONE);
-  ++g_on_resolution_invocations_count;
-  gpr_log(GPR_INFO,
-          "4th: g_on_resolution_invocations_count: %d, g_resolution_count: %d",
-          g_on_resolution_invocations_count, g_resolution_count);
-  // In this case we expect to have incurred in another system-level resolution
-  // because on_third_resolution slept for longer than the min resolution
-  // period.
-  GPR_ASSERT(g_on_resolution_invocations_count == 4);
-  GPR_ASSERT(g_resolution_count == 3);
+  gpr_log(GPR_INFO, "2nd: g_resolution_count: %d", g_resolution_count);
+  // The resolution callback was not invoked until new data was
+  // available, which was delayed until after the cooldown period.
+  GPR_ASSERT(g_resolution_count == 2);
   cb_arg->resolver.reset();
   gpr_atm_rel_store(&g_iomgr_args.done_atm, 1);
   gpr_mu_lock(g_iomgr_args.mu);
@@ -179,67 +193,13 @@ void on_fourth_resolution(void* arg, grpc_error* error) {
   g_all_callbacks_invoked = true;
 }
 
-void on_third_resolution(void* arg, grpc_error* error) {
-  OnResolutionCallbackArg* cb_arg = static_cast<OnResolutionCallbackArg*>(arg);
-  grpc_channel_args_destroy(cb_arg->result);
-  GPR_ASSERT(error == GRPC_ERROR_NONE);
-  ++g_on_resolution_invocations_count;
-  gpr_log(GPR_INFO,
-          "3rd: g_on_resolution_invocations_count: %d, g_resolution_count: %d",
-          g_on_resolution_invocations_count, g_resolution_count);
-  // The timer set because of the previous re-resolution request fires, so a new
-  // system-level resolution happened.
-  GPR_ASSERT(g_on_resolution_invocations_count == 3);
-  GPR_ASSERT(g_resolution_count == 2);
-  grpc_core::ExecCtx::Get()->TestOnlySetNow(
-      cb_arg->delay_before_second_resolution * 2);
-  cb_arg->resolver->NextLocked(
-      &cb_arg->result,
-      GRPC_CLOSURE_CREATE(on_fourth_resolution, arg,
-                          grpc_combiner_scheduler(g_combiner)));
-  cb_arg->resolver->RequestReresolutionLocked();
-  gpr_mu_lock(g_iomgr_args.mu);
-  GRPC_LOG_IF_ERROR("pollset_kick",
-                    grpc_pollset_kick(g_iomgr_args.pollset, nullptr));
-  gpr_mu_unlock(g_iomgr_args.mu);
-}
-
-void on_second_resolution(void* arg, grpc_error* error) {
-  OnResolutionCallbackArg* cb_arg = static_cast<OnResolutionCallbackArg*>(arg);
-  grpc_channel_args_destroy(cb_arg->result);
-  GPR_ASSERT(error == GRPC_ERROR_NONE);
-  ++g_on_resolution_invocations_count;
-  gpr_log(GPR_INFO,
-          "2nd: g_on_resolution_invocations_count: %d, g_resolution_count: %d",
-          g_on_resolution_invocations_count, g_resolution_count);
-  // The resolution request for which this function is the callback happened
-  // before the min resolution period. Therefore, no new system-level
-  // resolutions happened, as indicated by g_resolution_count. But a resolution
-  // timer was set to fire when the cooldown finishes.
-  GPR_ASSERT(g_on_resolution_invocations_count == 2);
-  GPR_ASSERT(g_resolution_count == 1);
-  // Register a new callback to capture the timer firing.
-  cb_arg->resolver->NextLocked(
-      &cb_arg->result,
-      GRPC_CLOSURE_CREATE(on_third_resolution, arg,
-                          grpc_combiner_scheduler(g_combiner)));
-  gpr_mu_lock(g_iomgr_args.mu);
-  GRPC_LOG_IF_ERROR("pollset_kick",
-                    grpc_pollset_kick(g_iomgr_args.pollset, nullptr));
-  gpr_mu_unlock(g_iomgr_args.mu);
-}
-
-void on_first_resolution(void* arg, grpc_error* error) {
+static void on_first_resolution(void* arg, grpc_error* error) {
   OnResolutionCallbackArg* cb_arg = static_cast<OnResolutionCallbackArg*>(arg);
   grpc_channel_args_destroy(cb_arg->result);
   GPR_ASSERT(error == GRPC_ERROR_NONE);
-  ++g_on_resolution_invocations_count;
-  gpr_log(GPR_INFO,
-          "1st: g_on_resolution_invocations_count: %d, g_resolution_count: %d",
-          g_on_resolution_invocations_count, g_resolution_count);
+  gpr_log(GPR_INFO, "1st: g_resolution_count: %d", g_resolution_count);
   // There's one initial system-level resolution and one invocation of a
   // notification callback (the current function).
-  GPR_ASSERT(g_on_resolution_invocations_count == 1);
   GPR_ASSERT(g_resolution_count == 1);
   cb_arg->resolver->NextLocked(
       &cb_arg->result,
@@ -265,9 +225,7 @@ static void start_test_under_combiner(void* arg, grpc_error* error) {
   grpc_core::ResolverArgs args;
   args.uri = uri;
   args.combiner = g_combiner;
-  g_on_resolution_invocations_count = 0;
   g_resolution_count = 0;
-  constexpr int kMinResolutionPeriodMs = 1000;
 
   grpc_arg cooldown_arg;
   cooldown_arg.key =
@@ -280,7 +238,6 @@ static void start_test_under_combiner(void* arg, grpc_error* error) {
   res_cb_arg->resolver = factory->CreateResolver(args);
   grpc_channel_args_destroy(cooldown_channel_args);
   GPR_ASSERT(res_cb_arg->resolver != nullptr);
-  res_cb_arg->delay_before_second_resolution = kMinResolutionPeriodMs;
   // First resolution, would incur in system-level resolution.
   res_cb_arg->resolver->NextLocked(
       &res_cb_arg->result,
diff --git a/test/core/client_channel/resolvers/fake_resolver_test.cc b/test/core/client_channel/resolvers/fake_resolver_test.cc
index 14caa3ea5d..f6696bf127 100644
--- a/test/core/client_channel/resolvers/fake_resolver_test.cc
+++ b/test/core/client_channel/resolvers/fake_resolver_test.cc
@@ -124,8 +124,8 @@ static void test_fake_resolver() {
       build_fake_resolver(combiner, response_generator.get());
   GPR_ASSERT(resolver.get() != nullptr);
   // Test 1: normal resolution.
-  // next_results != NULL, reresolution_results == NULL, last_used_results ==
-  // NULL. Expected response is next_results.
+  // next_results != NULL, reresolution_results == NULL.
+  // Expected response is next_results.
   grpc_channel_args* results = create_new_resolver_result();
   on_resolution_arg on_res_arg = create_on_resolution_arg(results);
   grpc_closure* on_resolution = GRPC_CLOSURE_CREATE(
@@ -137,10 +137,9 @@ static void test_fake_resolver() {
   GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
                             grpc_timeout_seconds_to_deadline(5)) != nullptr);
   // Test 2: update resolution.
-  // next_results != NULL, reresolution_results == NULL, last_used_results !=
-  // NULL. Expected response is next_results.
+  // next_results != NULL, reresolution_results == NULL.
+  // Expected response is next_results.
   results = create_new_resolver_result();
-  grpc_channel_args* last_used_results = grpc_channel_args_copy(results);
   on_res_arg = create_on_resolution_arg(results);
   on_resolution = GRPC_CLOSURE_CREATE(on_resolution_cb, &on_res_arg,
                                       grpc_combiner_scheduler(combiner));
@@ -150,21 +149,9 @@ static void test_fake_resolver() {
   grpc_core::ExecCtx::Get()->Flush();
   GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
                             grpc_timeout_seconds_to_deadline(5)) != nullptr);
-  // Test 3: fallback re-resolution.
-  // next_results == NULL, reresolution_results == NULL, last_used_results !=
-  // NULL. Expected response is last_used_results.
-  on_res_arg = create_on_resolution_arg(last_used_results);
-  on_resolution = GRPC_CLOSURE_CREATE(on_resolution_cb, &on_res_arg,
-                                      grpc_combiner_scheduler(combiner));
-  resolver->NextLocked(&on_res_arg.resolver_result, on_resolution);
-  // Trigger a re-resolution.
-  resolver->RequestReresolutionLocked();
-  grpc_core::ExecCtx::Get()->Flush();
-  GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
-                            grpc_timeout_seconds_to_deadline(5)) != nullptr);
-  // Test 4: normal re-resolution.
-  // next_results == NULL, reresolution_results != NULL, last_used_results !=
-  // NULL. Expected response is reresolution_results.
+  // Test 3: normal re-resolution.
+  // next_results == NULL, reresolution_results != NULL.
+  // Expected response is reresolution_results.
   grpc_channel_args* reresolution_results = create_new_resolver_result();
   on_res_arg =
       create_on_resolution_arg(grpc_channel_args_copy(reresolution_results));
@@ -180,9 +167,9 @@ static void test_fake_resolver() {
   grpc_core::ExecCtx::Get()->Flush();
   GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
                             grpc_timeout_seconds_to_deadline(5)) != nullptr);
-  // Test 5: repeat re-resolution.
-  // next_results == NULL, reresolution_results != NULL, last_used_results !=
-  // NULL. Expected response is reresolution_results.
+  // Test 4: repeat re-resolution.
+  // next_results == NULL, reresolution_results != NULL.
+  // Expected response is reresolution_results.
   on_res_arg = create_on_resolution_arg(reresolution_results);
   on_resolution = GRPC_CLOSURE_CREATE(on_resolution_cb, &on_res_arg,
                                       grpc_combiner_scheduler(combiner));
@@ -192,11 +179,10 @@ static void test_fake_resolver() {
   grpc_core::ExecCtx::Get()->Flush();
   GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
                             grpc_timeout_seconds_to_deadline(5)) != nullptr);
-  // Test 6: normal resolution.
-  // next_results != NULL, reresolution_results != NULL, last_used_results !=
-  // NULL. Expected response is next_results.
+  // Test 5: normal resolution.
+  // next_results != NULL, reresolution_results != NULL.
+  // Expected response is next_results.
   results = create_new_resolver_result();
-  last_used_results = grpc_channel_args_copy(results);
   on_res_arg = create_on_resolution_arg(results);
   on_resolution = GRPC_CLOSURE_CREATE(on_resolution_cb, &on_res_arg,
                                       grpc_combiner_scheduler(combiner));
@@ -206,23 +192,7 @@ static void test_fake_resolver() {
   grpc_core::ExecCtx::Get()->Flush();
   GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
                             grpc_timeout_seconds_to_deadline(5)) != nullptr);
-  // Test 7: fallback re-resolution.
-  // next_results == NULL, reresolution_results == NULL, last_used_results !=
-  // NULL. Expected response is last_used_results.
-  on_res_arg = create_on_resolution_arg(last_used_results);
-  on_resolution = GRPC_CLOSURE_CREATE(on_resolution_cb, &on_res_arg,
-                                      grpc_combiner_scheduler(combiner));
-  resolver->NextLocked(&on_res_arg.resolver_result, on_resolution);
-  // Reset reresolution_results.
-  response_generator->SetReresolutionResponse(nullptr);
-  // Flush here to guarantee that reresolution_results has been reset.
-  grpc_core::ExecCtx::Get()->Flush();
-  // Trigger a re-resolution.
-  resolver->RequestReresolutionLocked();
-  grpc_core::ExecCtx::Get()->Flush();
-  GPR_ASSERT(gpr_event_wait(&on_res_arg.ev,
-                            grpc_timeout_seconds_to_deadline(5)) != nullptr);
-  // Test 8: no-op.
+  // Test 6: no-op.
   // Requesting a new resolution without setting the response shouldn't trigger
   // the resolution callback.
   memset(&on_res_arg, 0, sizeof(on_res_arg));
diff --git a/test/core/end2end/BUILD b/test/core/end2end/BUILD
index dd16694204..37999a98d1 100644
--- a/test/core/end2end/BUILD
+++ b/test/core/end2end/BUILD
@@ -124,6 +124,19 @@ grpc_cc_test(
 )
 
 grpc_cc_test(
+    name = "inproc_callback_test",
+    srcs = ["inproc_callback_test.cc"],
+    language = "C++",
+    deps = [
+        ':end2end_tests',
+        "//:gpr",
+        "//:grpc",
+        "//test/core/util:gpr_test_util",
+        "//test/core/util:grpc_test_util",
+    ],
+)
+
+grpc_cc_test(
     name = "invalid_call_argument_test",
     srcs = ["invalid_call_argument_test.cc"],
     language = "C++",
diff --git a/test/core/end2end/bad_server_response_test.cc b/test/core/end2end/bad_server_response_test.cc
index 3d133cfc18..f7396a1684 100644
--- a/test/core/end2end/bad_server_response_test.cc
+++ b/test/core/end2end/bad_server_response_test.cc
@@ -104,7 +104,7 @@ static void handle_write() {
 
   grpc_slice_buffer_reset_and_unref(&state.outgoing_buffer);
   grpc_slice_buffer_add(&state.outgoing_buffer, slice);
-  grpc_endpoint_write(state.tcp, &state.outgoing_buffer, &on_write);
+  grpc_endpoint_write(state.tcp, &state.outgoing_buffer, &on_write, nullptr);
 }
 
 static void handle_read(void* arg, grpc_error* error) {
diff --git a/test/core/end2end/end2end_test.sh b/test/core/end2end/end2end_test.sh
index 5bfb253090..6b23d848be 100755
--- a/test/core/end2end/end2end_test.sh
+++ b/test/core/end2end/end2end_test.sh
@@ -15,7 +15,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-if [ -z "$3" ]
+if [ -n "$3" ]
   then
     export GRPC_POLL_STRATEGY=$3
 fi
diff --git a/test/core/end2end/fixtures/http_proxy_fixture.cc b/test/core/end2end/fixtures/http_proxy_fixture.cc
index f02fa9d998..ea9c000efb 100644
--- a/test/core/end2end/fixtures/http_proxy_fixture.cc
+++ b/test/core/end2end/fixtures/http_proxy_fixture.cc
@@ -201,7 +201,7 @@ static void on_client_write_done(void* arg, grpc_error* error) {
                                 &conn->client_write_buffer);
     conn->client_is_writing = true;
     grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer,
-                        &conn->on_client_write_done);
+                        &conn->on_client_write_done, nullptr);
   } else {
     // No more writes.  Unref the connection.
     proxy_connection_unref(conn, "write_done");
@@ -226,7 +226,7 @@ static void on_server_write_done(void* arg, grpc_error* error) {
                                 &conn->server_write_buffer);
     conn->server_is_writing = true;
     grpc_endpoint_write(conn->server_endpoint, &conn->server_write_buffer,
-                        &conn->on_server_write_done);
+                        &conn->on_server_write_done, nullptr);
   } else {
     // No more writes.  Unref the connection.
     proxy_connection_unref(conn, "server_write");
@@ -257,7 +257,7 @@ static void on_client_read_done(void* arg, grpc_error* error) {
     proxy_connection_ref(conn, "client_read");
     conn->server_is_writing = true;
     grpc_endpoint_write(conn->server_endpoint, &conn->server_write_buffer,
-                        &conn->on_server_write_done);
+                        &conn->on_server_write_done, nullptr);
   }
   // Read more data.
   grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer,
@@ -288,7 +288,7 @@ static void on_server_read_done(void* arg, grpc_error* error) {
     proxy_connection_ref(conn, "server_read");
     conn->client_is_writing = true;
     grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer,
-                        &conn->on_client_write_done);
+                        &conn->on_client_write_done, nullptr);
   }
   // Read more data.
   grpc_endpoint_read(conn->server_endpoint, &conn->server_read_buffer,
@@ -340,7 +340,7 @@ static void on_server_connect_done(void* arg, grpc_error* error) {
   grpc_slice_buffer_add(&conn->client_write_buffer, slice);
   conn->client_is_writing = true;
   grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer,
-                      &conn->on_write_response_done);
+                      &conn->on_write_response_done, nullptr);
 }
 
 /**
diff --git a/test/core/end2end/inproc_callback_test.cc b/test/core/end2end/inproc_callback_test.cc
new file mode 100644
index 0000000000..0d6c7c75a8
--- /dev/null
+++ b/test/core/end2end/inproc_callback_test.cc
@@ -0,0 +1,498 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/sync.h>
+
+#include "src/core/ext/transport/inproc/inproc_transport.h"
+#include "src/core/lib/surface/channel.h"
+#include "src/core/lib/surface/completion_queue.h"
+#include "src/core/lib/surface/server.h"
+#include "test/core/util/port.h"
+#include "test/core/util/test_config.h"
+
+typedef struct inproc_fixture_data {
+  bool dummy;  // reserved for future expansion. Struct can't be empty
+} inproc_fixture_data;
+
+namespace {
+template <typename F>
+class CQDeletingCallback : public grpc_core::CQCallbackInterface {
+ public:
+  explicit CQDeletingCallback(F f) : func_(f) {}
+  ~CQDeletingCallback() override {}
+  void Run(bool ok) override {
+    func_(ok);
+    grpc_core::Delete(this);
+  }
+
+ private:
+  F func_;
+};
+
+template <typename F>
+grpc_core::CQCallbackInterface* NewDeletingCallback(F f) {
+  return grpc_core::New<CQDeletingCallback<F>>(f);
+}
+
+class ShutdownCallback : public grpc_core::CQCallbackInterface {
+ public:
+  ShutdownCallback() : done_(false) {
+    gpr_mu_init(&mu_);
+    gpr_cv_init(&cv_);
+  }
+  ~ShutdownCallback() override {}
+  void Run(bool ok) override {
+    gpr_log(GPR_DEBUG, "CQ shutdown notification invoked");
+    gpr_mu_lock(&mu_);
+    done_ = true;
+    gpr_cv_broadcast(&cv_);
+    gpr_mu_unlock(&mu_);
+  }
+  // The Wait function waits for a specified amount of
+  // time for the completion of the shutdown and returns
+  // whether it was successfully shut down
+  bool Wait(gpr_timespec deadline) {
+    gpr_mu_lock(&mu_);
+    while (!done_ && !gpr_cv_wait(&cv_, &mu_, deadline)) {
+    }
+    bool ret = done_;
+    gpr_mu_unlock(&mu_);
+    return ret;
+  }
+
+ private:
+  bool done_;
+  gpr_mu mu_;
+  gpr_cv cv_;
+};
+
+ShutdownCallback* g_shutdown_callback;
+}  // namespace
+
+// The following global structure is the tag collection. It holds
+// all information related to tags expected and tags received
+// during the execution, with each callback setting a tag.
+// The tag sets are implemented and checked using arrays and
+// linear lookups (rather than maps) so that this test doesn't
+// need the C++ standard library.
+static gpr_mu tags_mu;
+static gpr_cv tags_cv;
+const size_t kAvailableTags = 4;
+bool tags[kAvailableTags];
+bool tags_valid[kAvailableTags];
+bool tags_expected[kAvailableTags];
+bool tags_needed[kAvailableTags];
+
+// Mark that a tag is expected; this function must be executed in the
+// main thread only while there are no other threads altering the
+// expectation set (e.g., by calling expect_tag or verify_tags)
+static void expect_tag(intptr_t tag, bool ok) {
+  size_t idx = static_cast<size_t>(tag);
+  GPR_ASSERT(idx < kAvailableTags);
+  tags_needed[idx] = true;
+  tags_expected[idx] = ok;
+}
+
+// Check that the expected tags have reached, within a certain
+// deadline. This must also be executed only on the main thread while
+// there are no other threads altering the expectation set (e.g., by
+// calling expect_tag or verify_tags). The tag verifier doesn't have
+// to drive the CQ at all (unlike the next-based end2end tests)
+// because the tags will get set when the callbacks are executed,
+// which happens when a particular batch related to a callback is
+// complete.
+static void verify_tags(gpr_timespec deadline) {
+  bool done = false;
+
+  gpr_mu_lock(&tags_mu);
+  while (!done) {
+    done = gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), deadline) > 0;
+    for (size_t i = 0; i < kAvailableTags; i++) {
+      if (tags_needed[i]) {
+        if (tags_valid[i]) {
+          gpr_log(GPR_DEBUG, "Verifying tag %d", static_cast<int>(i));
+          if (tags[i] != tags_expected[i]) {
+            gpr_log(GPR_ERROR, "Got wrong result (%d instead of %d) for tag %d",
+                    tags[i], tags_expected[i], static_cast<int>(i));
+            GPR_ASSERT(false);
+          }
+          tags_valid[i] = false;
+          tags_needed[i] = false;
+        } else if (done) {
+          gpr_log(GPR_ERROR, "Didn't get tag %d", static_cast<int>(i));
+          GPR_ASSERT(false);
+        }
+      }
+    }
+    bool empty = true;
+    for (size_t i = 0; i < kAvailableTags; i++) {
+      if (tags_needed[i]) {
+        empty = false;
+      }
+    }
+    done = done || empty;
+    if (done) {
+      for (size_t i = 0; i < kAvailableTags; i++) {
+        if (tags_valid[i]) {
+          gpr_log(GPR_ERROR, "Got unexpected tag %d and result %d",
+                  static_cast<int>(i), tags[i]);
+          GPR_ASSERT(false);
+        }
+        tags_valid[i] = false;
+      }
+    } else {
+      gpr_cv_wait(&tags_cv, &tags_mu, deadline);
+    }
+  }
+  gpr_mu_unlock(&tags_mu);
+}
+
+// This function creates a callback functor that emits the
+// desired tag into the global tag set
+static grpc_core::CQCallbackInterface* tag(intptr_t t) {
+  auto func = [t](bool ok) {
+    gpr_mu_lock(&tags_mu);
+    gpr_log(GPR_DEBUG, "Completing operation %" PRIdPTR, t);
+    bool was_empty = true;
+    for (size_t i = 0; i < kAvailableTags; i++) {
+      if (tags_valid[i]) {
+        was_empty = false;
+      }
+    }
+    size_t idx = static_cast<size_t>(t);
+    tags[idx] = ok;
+    tags_valid[idx] = true;
+    if (was_empty) {
+      gpr_cv_signal(&tags_cv);
+    }
+    gpr_mu_unlock(&tags_mu);
+  };
+  auto cb = NewDeletingCallback(func);
+  return cb;
+}
+
+static grpc_end2end_test_fixture inproc_create_fixture(
+    grpc_channel_args* client_args, grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  inproc_fixture_data* ffd = static_cast<inproc_fixture_data*>(
+      gpr_malloc(sizeof(inproc_fixture_data)));
+  memset(&f, 0, sizeof(f));
+
+  f.fixture_data = ffd;
+  g_shutdown_callback = grpc_core::New<ShutdownCallback>();
+  f.cq =
+      grpc_completion_queue_create_for_callback(g_shutdown_callback, nullptr);
+  f.shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
+
+  return f;
+}
+
+void inproc_init_client(grpc_end2end_test_fixture* f,
+                        grpc_channel_args* client_args) {
+  f->client = grpc_inproc_channel_create(f->server, client_args, nullptr);
+  GPR_ASSERT(f->client);
+}
+
+void inproc_init_server(grpc_end2end_test_fixture* f,
+                        grpc_channel_args* server_args) {
+  if (f->server) {
+    grpc_server_destroy(f->server);
+  }
+  f->server = grpc_server_create(server_args, nullptr);
+  grpc_server_register_completion_queue(f->server, f->cq, nullptr);
+  grpc_server_start(f->server);
+}
+
+void inproc_tear_down(grpc_end2end_test_fixture* f) {
+  inproc_fixture_data* ffd = static_cast<inproc_fixture_data*>(f->fixture_data);
+  gpr_free(ffd);
+}
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char* test_name,
+                                            grpc_channel_args* client_args,
+                                            grpc_channel_args* server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_from_now(int n) {
+  return grpc_timeout_seconds_to_deadline(n);
+}
+
+static gpr_timespec five_seconds_from_now() { return n_seconds_from_now(5); }
+
+static void drain_cq(grpc_completion_queue* cq) {
+  // Wait for the shutdown callback to arrive, or fail the test
+  GPR_ASSERT(g_shutdown_callback->Wait(five_seconds_from_now()));
+  gpr_log(GPR_DEBUG, "CQ shutdown wait complete");
+  grpc_core::Delete(g_shutdown_callback);
+}
+
+static void shutdown_server(grpc_end2end_test_fixture* f) {
+  if (!f->server) return;
+  grpc_server_shutdown_and_notify(
+      f->server, f->shutdown_cq,
+      reinterpret_cast<void*>(static_cast<intptr_t>(1000)));
+  GPR_ASSERT(
+      grpc_completion_queue_pluck(f->shutdown_cq, (void*)((intptr_t)1000),
+                                  grpc_timeout_seconds_to_deadline(5), nullptr)
+          .type == GRPC_OP_COMPLETE);
+  grpc_server_destroy(f->server);
+  f->server = nullptr;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture* f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = nullptr;
+}
+
+static void end_test(grpc_end2end_test_fixture* f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+  grpc_completion_queue_destroy(f->shutdown_cq);
+}
+
+static void simple_request_body(grpc_end2end_test_config config,
+                                grpc_end2end_test_fixture f) {
+  grpc_call* c;
+  grpc_call* s;
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_call_details call_details;
+  grpc_status_code status;
+  const char* error_string;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+  char* peer;
+  gpr_timespec deadline = five_seconds_from_now();
+
+  c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+                               grpc_slice_from_static_string("/foo"), nullptr,
+                               deadline, nullptr);
+  GPR_ASSERT(c);
+
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
+  gpr_free(peer);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+
+  // Create a basic client unary request batch (no payload)
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op->data.recv_status_on_client.error_string = &error_string;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
+                                nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // Register a call at the server-side to match the incoming client call
+  error = grpc_server_request_call(f.server, &s, &call_details,
+                                   &request_metadata_recv, f.cq, f.cq, tag(2));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // We expect that the server call creation callback (and no others) will
+  // execute now since no other batch should be complete.
+  expect_tag(2, true);
+  verify_tags(deadline);
+
+  peer = grpc_call_get_peer(s);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "server_peer=%s", peer);
+  gpr_free(peer);
+  peer = grpc_call_get_peer(c);
+  GPR_ASSERT(peer != nullptr);
+  gpr_log(GPR_DEBUG, "client_peer=%s", peer);
+  gpr_free(peer);
+
+  // Create the server response batch (no payload)
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED;
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+  op->data.send_status_from_server.status_details = &status_details;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(3),
+                                nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  // Both the client request and server response batches should get complete
+  // now and we should see that their callbacks have been executed
+  expect_tag(3, true);
+  expect_tag(1, true);
+  verify_tags(deadline);
+
+  GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  // the following sanity check makes sure that the requested error string is
+  // correctly populated by the core. It looks for certain substrings that are
+  // not likely to change much. Some parts of the error, like time created,
+  // obviously are not checked.
+  GPR_ASSERT(nullptr != strstr(error_string, "xyz"));
+  GPR_ASSERT(nullptr != strstr(error_string, "description"));
+  GPR_ASSERT(nullptr != strstr(error_string, "Error received from peer"));
+  GPR_ASSERT(nullptr != strstr(error_string, "grpc_message"));
+  GPR_ASSERT(nullptr != strstr(error_string, "grpc_status"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
+  GPR_ASSERT(0 == call_details.flags);
+  GPR_ASSERT(was_cancelled == 1);
+
+  grpc_slice_unref(details);
+  gpr_free(static_cast<void*>(const_cast<char*>(error_string)));
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  int expected_calls = 1;
+  if (config.feature_mask & FEATURE_MASK_SUPPORTS_REQUEST_PROXYING) {
+    expected_calls *= 2;
+  }
+}
+
+static void test_invoke_simple_request(grpc_end2end_test_config config) {
+  grpc_end2end_test_fixture f;
+
+  f = begin_test(config, "test_invoke_simple_request", nullptr, nullptr);
+  simple_request_body(config, f);
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+static void test_invoke_10_simple_requests(grpc_end2end_test_config config) {
+  int i;
+  grpc_end2end_test_fixture f =
+      begin_test(config, "test_invoke_10_simple_requests", nullptr, nullptr);
+  for (i = 0; i < 10; i++) {
+    simple_request_body(config, f);
+    gpr_log(GPR_INFO, "Running test: Passed simple request %d", i);
+  }
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+static void test_invoke_many_simple_requests(grpc_end2end_test_config config) {
+  int i;
+  const int many = 1000;
+  grpc_end2end_test_fixture f =
+      begin_test(config, "test_invoke_many_simple_requests", nullptr, nullptr);
+  gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
+  for (i = 0; i < many; i++) {
+    simple_request_body(config, f);
+  }
+  double us =
+      gpr_timespec_to_micros(gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), t1)) /
+      many;
+  gpr_log(GPR_INFO, "Time per ping %f us", us);
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+static void simple_request(grpc_end2end_test_config config) {
+  int i;
+  for (i = 0; i < 10; i++) {
+    test_invoke_simple_request(config);
+  }
+  test_invoke_10_simple_requests(config);
+  test_invoke_many_simple_requests(config);
+}
+
+static void simple_request_pre_init() {
+  gpr_mu_init(&tags_mu);
+  gpr_cv_init(&tags_cv);
+}
+
+/* All test configurations */
+static grpc_end2end_test_config configs[] = {
+    {"inproc-callback", FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, nullptr,
+     inproc_create_fixture, inproc_init_client, inproc_init_server,
+     inproc_tear_down},
+};
+
+int main(int argc, char** argv) {
+  grpc_test_init(argc, argv);
+  grpc_init();
+
+  simple_request_pre_init();
+  simple_request(configs[0]);
+
+  grpc_shutdown();
+
+  return 0;
+}
diff --git a/test/core/end2end/tests/filter_call_init_fails.cc b/test/core/end2end/tests/filter_call_init_fails.cc
index 07e1421446..ab96879fe4 100644
--- a/test/core/end2end/tests/filter_call_init_fails.cc
+++ b/test/core/end2end/tests/filter_call_init_fails.cc
@@ -438,6 +438,7 @@ static bool maybe_add_server_channel_filter(grpc_channel_stack_builder* builder,
     // must be the last one.  So we add it right before the last one.
     grpc_channel_stack_builder_iterator* it =
         grpc_channel_stack_builder_create_iterator_at_last(builder);
+    GPR_ASSERT(grpc_channel_stack_builder_move_prev(it));
     const bool retval = grpc_channel_stack_builder_add_filter_before(
         it, &test_filter, nullptr, nullptr);
     grpc_channel_stack_builder_iterator_destroy(it);
@@ -456,6 +457,7 @@ static bool maybe_add_client_channel_filter(grpc_channel_stack_builder* builder,
     // must be the last one.  So we add it right before the last one.
     grpc_channel_stack_builder_iterator* it =
         grpc_channel_stack_builder_create_iterator_at_last(builder);
+    GPR_ASSERT(grpc_channel_stack_builder_move_prev(it));
     const bool retval = grpc_channel_stack_builder_add_filter_before(
         it, &test_filter, nullptr, nullptr);
     grpc_channel_stack_builder_iterator_destroy(it);
@@ -474,6 +476,7 @@ static bool maybe_add_client_subchannel_filter(
     // must be the last one.  So we add it right before the last one.
     grpc_channel_stack_builder_iterator* it =
         grpc_channel_stack_builder_create_iterator_at_last(builder);
+    GPR_ASSERT(grpc_channel_stack_builder_move_prev(it));
     const bool retval = grpc_channel_stack_builder_add_filter_before(
         it, &test_filter, nullptr, nullptr);
     grpc_channel_stack_builder_iterator_destroy(it);
@@ -484,17 +487,13 @@ static bool maybe_add_client_subchannel_filter(
 }
 
 static void init_plugin(void) {
-  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL,
-                                   GRPC_CHANNEL_INIT_PRIORITY_MAX,
+  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX,
                                    maybe_add_server_channel_filter, nullptr);
-  grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL,
-                                   GRPC_CHANNEL_INIT_PRIORITY_MAX,
+  grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX,
                                    maybe_add_client_channel_filter, nullptr);
-  grpc_channel_init_register_stage(GRPC_CLIENT_SUBCHANNEL,
-                                   GRPC_CHANNEL_INIT_PRIORITY_MAX,
+  grpc_channel_init_register_stage(GRPC_CLIENT_SUBCHANNEL, INT_MAX,
                                    maybe_add_client_subchannel_filter, nullptr);
-  grpc_channel_init_register_stage(GRPC_CLIENT_DIRECT_CHANNEL,
-                                   GRPC_CHANNEL_INIT_PRIORITY_MAX,
+  grpc_channel_init_register_stage(GRPC_CLIENT_DIRECT_CHANNEL, INT_MAX,
                                    maybe_add_client_channel_filter, nullptr);
 }
 
diff --git a/test/core/end2end/tests/filter_causes_close.cc b/test/core/end2end/tests/filter_causes_close.cc
index 891c1b8c1f..a7f4268803 100644
--- a/test/core/end2end/tests/filter_causes_close.cc
+++ b/test/core/end2end/tests/filter_causes_close.cc
@@ -261,9 +261,8 @@ static bool maybe_add_filter(grpc_channel_stack_builder* builder, void* arg) {
 }
 
 static void init_plugin(void) {
-  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL,
-                                   GRPC_CHANNEL_INIT_PRIORITY_HIGH,
-                                   maybe_add_filter, nullptr);
+  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, 0, maybe_add_filter,
+                                   nullptr);
 }
 
 static void destroy_plugin(void) {}
diff --git a/test/core/end2end/tests/filter_latency.cc b/test/core/end2end/tests/filter_latency.cc
index 02a4d07927..a89db7b094 100644
--- a/test/core/end2end/tests/filter_latency.cc
+++ b/test/core/end2end/tests/filter_latency.cc
@@ -314,6 +314,7 @@ static bool maybe_add_filter(grpc_channel_stack_builder* builder, void* arg) {
     // must be the last one.  So we add it right before the last one.
     grpc_channel_stack_builder_iterator* it =
         grpc_channel_stack_builder_create_iterator_at_last(builder);
+    GPR_ASSERT(grpc_channel_stack_builder_move_prev(it));
     const bool retval = grpc_channel_stack_builder_add_filter_before(
         it, filter, nullptr, nullptr);
     grpc_channel_stack_builder_iterator_destroy(it);
@@ -325,15 +326,15 @@ static bool maybe_add_filter(grpc_channel_stack_builder* builder, void* arg) {
 
 static void init_plugin(void) {
   gpr_mu_init(&g_mu);
-  grpc_channel_init_register_stage(
-      GRPC_CLIENT_CHANNEL, GRPC_CHANNEL_INIT_PRIORITY_MAX, maybe_add_filter,
-      (void*)&test_client_filter);
-  grpc_channel_init_register_stage(
-      GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_PRIORITY_MAX,
-      maybe_add_filter, (void*)&test_client_filter);
-  grpc_channel_init_register_stage(
-      GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_PRIORITY_MAX, maybe_add_filter,
-      (void*)&test_server_filter);
+  grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX,
+                                   maybe_add_filter,
+                                   (void*)&test_client_filter);
+  grpc_channel_init_register_stage(GRPC_CLIENT_DIRECT_CHANNEL, INT_MAX,
+                                   maybe_add_filter,
+                                   (void*)&test_client_filter);
+  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX,
+                                   maybe_add_filter,
+                                   (void*)&test_server_filter);
 }
 
 static void destroy_plugin(void) { gpr_mu_destroy(&g_mu); }
diff --git a/test/core/end2end/tests/filter_status_code.cc b/test/core/end2end/tests/filter_status_code.cc
index 6ed1de15c6..ba3cbfa6d1 100644
--- a/test/core/end2end/tests/filter_status_code.cc
+++ b/test/core/end2end/tests/filter_status_code.cc
@@ -333,6 +333,7 @@ static bool maybe_add_filter(grpc_channel_stack_builder* builder, void* arg) {
     // So we add it right before the last one.
     grpc_channel_stack_builder_iterator* it =
         grpc_channel_stack_builder_create_iterator_at_last(builder);
+    GPR_ASSERT(grpc_channel_stack_builder_move_prev(it));
     const bool retval = grpc_channel_stack_builder_add_filter_before(
         it, filter, nullptr, nullptr);
     grpc_channel_stack_builder_iterator_destroy(it);
@@ -349,15 +350,15 @@ static void init_plugin(void) {
   g_client_code_recv = false;
   g_server_code_recv = false;
 
-  grpc_channel_init_register_stage(
-      GRPC_CLIENT_CHANNEL, GRPC_CHANNEL_INIT_PRIORITY_MAX, maybe_add_filter,
-      (void*)&test_client_filter);
-  grpc_channel_init_register_stage(
-      GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_PRIORITY_MAX,
-      maybe_add_filter, (void*)&test_client_filter);
-  grpc_channel_init_register_stage(
-      GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_PRIORITY_MAX, maybe_add_filter,
-      (void*)&test_server_filter);
+  grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX,
+                                   maybe_add_filter,
+                                   (void*)&test_client_filter);
+  grpc_channel_init_register_stage(GRPC_CLIENT_DIRECT_CHANNEL, INT_MAX,
+                                   maybe_add_filter,
+                                   (void*)&test_client_filter);
+  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX,
+                                   maybe_add_filter,
+                                   (void*)&test_server_filter);
 }
 
 static void destroy_plugin(void) {
diff --git a/test/core/iomgr/BUILD b/test/core/iomgr/BUILD
index 002671a5fa..675d9e6278 100644
--- a/test/core/iomgr/BUILD
+++ b/test/core/iomgr/BUILD
@@ -247,6 +247,19 @@ grpc_cc_test(
 )
 
 grpc_cc_test(
+    name = "buffer_list_test",
+    srcs = ["buffer_list_test.cc"],
+    language = "C++",
+    deps = [
+        "//:gpr",
+        "//:grpc",
+        "//test/core/util:gpr_test_util",
+        "//test/core/util:grpc_test_util",
+    ],
+)
+
+
+grpc_cc_test(
     name = "tcp_server_posix_test",
     srcs = ["tcp_server_posix_test.cc"],
     language = "C++",
diff --git a/test/core/iomgr/buffer_list_test.cc b/test/core/iomgr/buffer_list_test.cc
new file mode 100644
index 0000000000..c7f30fa092
--- /dev/null
+++ b/test/core/iomgr/buffer_list_test.cc
@@ -0,0 +1,111 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "src/core/lib/iomgr/port.h"
+
+#include "src/core/lib/iomgr/buffer_list.h"
+
+#include <grpc/grpc.h>
+
+#include "test/core/util/test_config.h"
+
+#ifdef GRPC_LINUX_ERRQUEUE
+
+static void TestShutdownFlushesListVerifier(void* arg,
+                                            grpc_core::Timestamps* ts,
+                                            grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
+  GPR_ASSERT(arg != nullptr);
+  gpr_atm* done = reinterpret_cast<gpr_atm*>(arg);
+  gpr_atm_rel_store(done, static_cast<gpr_atm>(1));
+}
+
+/** Tests that all TracedBuffer elements in the list are flushed out on
+ * shutdown.
+ * Also tests that arg is passed correctly.
+ */
+static void TestShutdownFlushesList() {
+  grpc_core::grpc_tcp_set_write_timestamps_callback(
+      TestShutdownFlushesListVerifier);
+  grpc_core::TracedBuffer* list = nullptr;
+#define NUM_ELEM 5
+  gpr_atm verifier_called[NUM_ELEM];
+  for (auto i = 0; i < NUM_ELEM; i++) {
+    gpr_atm_rel_store(&verifier_called[i], static_cast<gpr_atm>(0));
+    grpc_core::TracedBuffer::AddNewEntry(
+        &list, i, static_cast<void*>(&verifier_called[i]));
+  }
+  grpc_core::TracedBuffer::Shutdown(&list, GRPC_ERROR_NONE);
+  GPR_ASSERT(list == nullptr);
+  for (auto i = 0; i < NUM_ELEM; i++) {
+    GPR_ASSERT(gpr_atm_acq_load(&verifier_called[i]) ==
+               static_cast<gpr_atm>(1));
+  }
+}
+
+static void TestVerifierCalledOnAckVerifier(void* arg,
+                                            grpc_core::Timestamps* ts,
+                                            grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
+  GPR_ASSERT(arg != nullptr);
+  GPR_ASSERT(ts->acked_time.clock_type == GPR_CLOCK_REALTIME);
+  GPR_ASSERT(ts->acked_time.tv_sec == 123);
+  GPR_ASSERT(ts->acked_time.tv_nsec == 456);
+  gpr_atm* done = reinterpret_cast<gpr_atm*>(arg);
+  gpr_atm_rel_store(done, static_cast<gpr_atm>(1));
+}
+
+/** Tests that the timestamp verifier is called on an ACK timestamp.
+ */
+static void TestVerifierCalledOnAck() {
+  struct sock_extended_err serr;
+  serr.ee_data = 213;
+  serr.ee_info = grpc_core::SCM_TSTAMP_ACK;
+  struct grpc_core::scm_timestamping tss;
+  tss.ts[0].tv_sec = 123;
+  tss.ts[0].tv_nsec = 456;
+  grpc_core::grpc_tcp_set_write_timestamps_callback(
+      TestVerifierCalledOnAckVerifier);
+  grpc_core::TracedBuffer* list = nullptr;
+  gpr_atm verifier_called;
+  gpr_atm_rel_store(&verifier_called, static_cast<gpr_atm>(0));
+  grpc_core::TracedBuffer::AddNewEntry(&list, 213, &verifier_called);
+  grpc_core::TracedBuffer::ProcessTimestamp(&list, &serr, &tss);
+  GPR_ASSERT(gpr_atm_acq_load(&verifier_called) == static_cast<gpr_atm>(1));
+  GPR_ASSERT(list == nullptr);
+  grpc_core::TracedBuffer::Shutdown(&list, GRPC_ERROR_NONE);
+}
+
+static void TestTcpBufferList() {
+  TestVerifierCalledOnAck();
+  TestShutdownFlushesList();
+}
+
+int main(int argc, char** argv) {
+  grpc_test_init(argc, argv);
+  grpc_init();
+  TestTcpBufferList();
+  grpc_shutdown();
+  return 0;
+}
+
+#else /* GRPC_LINUX_ERRQUEUE */
+
+int main(int argc, char** argv) { return 0; }
+
+#endif /* GRPC_LINUX_ERRQUEUE */
diff --git a/test/core/iomgr/endpoint_tests.cc b/test/core/iomgr/endpoint_tests.cc
index 8db8ac5ed6..a9e8ba86c5 100644
--- a/test/core/iomgr/endpoint_tests.cc
+++ b/test/core/iomgr/endpoint_tests.cc
@@ -150,8 +150,8 @@ static void read_and_write_test_write_handler(void* data, grpc_error* error) {
                                &state->current_write_data);
       grpc_slice_buffer_reset_and_unref(&state->outgoing);
       grpc_slice_buffer_addn(&state->outgoing, slices, nslices);
-      grpc_endpoint_write(state->write_ep, &state->outgoing,
-                          &state->done_write);
+      grpc_endpoint_write(state->write_ep, &state->outgoing, &state->done_write,
+                          nullptr);
       gpr_free(slices);
       return;
     }
@@ -294,7 +294,8 @@ static void multiple_shutdown_test(grpc_endpoint_test_config config) {
   grpc_slice_buffer_add(&slice_buffer, grpc_slice_from_copied_string("a"));
   grpc_endpoint_write(f.client_ep, &slice_buffer,
                       GRPC_CLOSURE_CREATE(inc_on_failure, &fail_count,
-                                          grpc_schedule_on_exec_ctx));
+                                          grpc_schedule_on_exec_ctx),
+                      nullptr);
   wait_for_fail_count(&fail_count, 3);
   grpc_endpoint_shutdown(f.client_ep,
                          GRPC_ERROR_CREATE_FROM_STATIC_STRING("Test Shutdown"));
diff --git a/test/core/iomgr/ios/CFStreamTests/Podfile b/test/core/iomgr/ios/CFStreamTests/Podfile
index 630168a363..e6ec66d549 100644
--- a/test/core/iomgr/ios/CFStreamTests/Podfile
+++ b/test/core/iomgr/ios/CFStreamTests/Podfile
@@ -9,6 +9,7 @@ GRPC_LOCAL_SRC = '../../../../..'
 # Install the dependencies in the main target plus all test targets.
 target 'CFStreamTests' do
   pod 'gRPC-Core/CFStream-Implementation', :path => GRPC_LOCAL_SRC
+  pod 'BoringSSL-GRPC', :podspec => "#{GRPC_LOCAL_SRC}/src/objective-c", :inhibit_warnings => true
 end
 
 pre_install do |installer|
diff --git a/test/core/iomgr/resource_quota_test.cc b/test/core/iomgr/resource_quota_test.cc
index 059ff7b5f8..f3b35fed32 100644
--- a/test/core/iomgr/resource_quota_test.cc
+++ b/test/core/iomgr/resource_quota_test.cc
@@ -798,6 +798,98 @@ static void test_negative_rq_free_pool(void) {
   }
 }
 
+// Simple test to check resource quota thread limits
+static void test_thread_limit() {
+  grpc_core::ExecCtx exec_ctx;
+
+  grpc_resource_quota* rq = grpc_resource_quota_create("test_thread_limit");
+  grpc_resource_user* ru1 = grpc_resource_user_create(rq, "ru1");
+  grpc_resource_user* ru2 = grpc_resource_user_create(rq, "ru2");
+
+  // Max threads = 100
+  grpc_resource_quota_set_max_threads(rq, 100);
+
+  // Request quota for 100 threads (50 for ru1, 50 for ru2)
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru1, 10));
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 10));
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru1, 40));
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 40));
+
+  // Threads exhausted. Next request must fail
+  GPR_ASSERT(!grpc_resource_user_allocate_threads(ru2, 20));
+
+  // Free 20 threads from two different users
+  grpc_resource_user_free_threads(ru1, 10);
+  grpc_resource_user_free_threads(ru2, 10);
+
+  // Next request to 20 threads must succeed
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 20));
+
+  // No more thread quota again
+  GPR_ASSERT(!grpc_resource_user_allocate_threads(ru1, 20));
+
+  // Free 10 more
+  grpc_resource_user_free_threads(ru1, 10);
+
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru1, 5));
+  GPR_ASSERT(
+      !grpc_resource_user_allocate_threads(ru2, 10));  // Only 5 available
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 5));
+
+  // Teardown (ru1 and ru2 release all the quota back to rq)
+  grpc_resource_user_unref(ru1);
+  grpc_resource_user_unref(ru2);
+  grpc_resource_quota_unref(rq);
+}
+
+// Change max quota in either direction dynamically
+static void test_thread_maxquota_change() {
+  grpc_core::ExecCtx exec_ctx;
+
+  grpc_resource_quota* rq =
+      grpc_resource_quota_create("test_thread_maxquota_change");
+  grpc_resource_user* ru1 = grpc_resource_user_create(rq, "ru1");
+  grpc_resource_user* ru2 = grpc_resource_user_create(rq, "ru2");
+
+  // Max threads = 100
+  grpc_resource_quota_set_max_threads(rq, 100);
+
+  // Request quota for 100 threads (50 for ru1, 50 for ru2)
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru1, 50));
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 50));
+
+  // Threads exhausted. Next request must fail
+  GPR_ASSERT(!grpc_resource_user_allocate_threads(ru2, 20));
+
+  // Increase maxquota and retry
+  // Max threads = 150;
+  grpc_resource_quota_set_max_threads(rq, 150);
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 20));  // ru2=70, ru1=50
+
+  // Decrease maxquota (Note: Quota already given to ru1 and ru2 is unaffected)
+  // Max threads = 10;
+  grpc_resource_quota_set_max_threads(rq, 10);
+
+  // New requests will fail until quota is available
+  GPR_ASSERT(!grpc_resource_user_allocate_threads(ru1, 10));
+
+  // Make quota available
+  grpc_resource_user_free_threads(ru1, 50);                   // ru1 now has 0
+  GPR_ASSERT(!grpc_resource_user_allocate_threads(ru1, 10));  // not enough
+
+  grpc_resource_user_free_threads(ru2, 70);  // ru2 now has 0
+
+  // Now we can get quota up-to 10, the current max
+  GPR_ASSERT(grpc_resource_user_allocate_threads(ru2, 10));
+  // No more thread quota again
+  GPR_ASSERT(!grpc_resource_user_allocate_threads(ru1, 10));
+
+  // Teardown (ru1 and ru2 release all the quota back to rq)
+  grpc_resource_user_unref(ru1);
+  grpc_resource_user_unref(ru2);
+  grpc_resource_quota_unref(rq);
+}
+
 int main(int argc, char** argv) {
   grpc_test_init(argc, argv);
   grpc_init();
@@ -827,6 +919,11 @@ int main(int argc, char** argv) {
   test_negative_rq_free_pool();
   gpr_mu_destroy(&g_mu);
   gpr_cv_destroy(&g_cv);
+
+  // Resource quota thread related
+  test_thread_limit();
+  test_thread_maxquota_change();
+
   grpc_shutdown();
   return 0;
 }
diff --git a/test/core/iomgr/tcp_posix_test.cc b/test/core/iomgr/tcp_posix_test.cc
index 3e87831e44..6447cc234d 100644
--- a/test/core/iomgr/tcp_posix_test.cc
+++ b/test/core/iomgr/tcp_posix_test.cc
@@ -36,6 +36,9 @@
 #include <grpc/support/time.h>
 
 #include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/iomgr/buffer_list.h"
+#include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/sockaddr_posix.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "test/core/iomgr/endpoint_tests.h"
 #include "test/core/util/test_config.h"
@@ -68,6 +71,43 @@ static void create_sockets(int sv[2]) {
   GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
 }
 
+static void create_inet_sockets(int sv[2]) {
+  /* Prepare listening socket */
+  struct sockaddr_in addr;
+  memset(&addr, 0, sizeof(struct sockaddr_in));
+  addr.sin_family = AF_INET;
+  int sock = socket(AF_INET, SOCK_STREAM, 0);
+  GPR_ASSERT(sock);
+  GPR_ASSERT(bind(sock, (sockaddr*)&addr, sizeof(sockaddr_in)) == 0);
+  listen(sock, 1);
+
+  /* Prepare client socket and connect to server */
+  socklen_t len = sizeof(sockaddr_in);
+  GPR_ASSERT(getsockname(sock, (sockaddr*)&addr, &len) == 0);
+
+  int client = socket(AF_INET, SOCK_STREAM, 0);
+  GPR_ASSERT(client);
+  int ret;
+  do {
+    ret = connect(client, (sockaddr*)&addr, sizeof(sockaddr_in));
+  } while (ret == -1 && errno == EINTR);
+
+  /* Accept client connection */
+  len = sizeof(socklen_t);
+  int server;
+  do {
+    server = accept(sock, (sockaddr*)&addr, (socklen_t*)&len);
+  } while (server == -1 && errno == EINTR);
+  GPR_ASSERT(server != -1);
+
+  sv[0] = server;
+  sv[1] = client;
+  int flags = fcntl(sv[0], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[0], F_SETFL, flags | O_NONBLOCK) == 0);
+  flags = fcntl(sv[1], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
+}
+
 static ssize_t fill_socket(int fd) {
   ssize_t write_bytes;
   ssize_t total_bytes = 0;
@@ -289,11 +329,10 @@ static grpc_slice* allocate_blocks(size_t num_bytes, size_t slice_size,
 
 static void write_done(void* user_data /* write_socket_state */,
                        grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
   struct write_socket_state* state =
       static_cast<struct write_socket_state*>(user_data);
-  gpr_log(GPR_INFO, "Write done callback called");
   gpr_mu_lock(g_mu);
-  gpr_log(GPR_INFO, "Signalling write done");
   state->write_done = 1;
   GPR_ASSERT(
       GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, nullptr)));
@@ -340,10 +379,24 @@ void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
   gpr_free(buf);
 }
 
+/* Verifier for timestamps callback for write_test */
+void timestamps_verifier(void* arg, grpc_core::Timestamps* ts,
+                         grpc_error* error) {
+  GPR_ASSERT(error == GRPC_ERROR_NONE);
+  GPR_ASSERT(arg != nullptr);
+  GPR_ASSERT(ts->sendmsg_time.clock_type == GPR_CLOCK_REALTIME);
+  GPR_ASSERT(ts->scheduled_time.clock_type == GPR_CLOCK_REALTIME);
+  GPR_ASSERT(ts->acked_time.clock_type == GPR_CLOCK_REALTIME);
+  gpr_atm* done_timestamps = (gpr_atm*)arg;
+  gpr_atm_rel_store(done_timestamps, static_cast<gpr_atm>(1));
+}
+
 /* Write to a socket using the grpc_tcp API, then drain it directly.
    Note that if the write does not complete immediately we need to drain the
-   socket in parallel with the read. */
-static void write_test(size_t num_bytes, size_t slice_size) {
+   socket in parallel with the read. If collect_timestamps is true, it will
+   try to get timestamps for the write. */
+static void write_test(size_t num_bytes, size_t slice_size,
+                       bool collect_timestamps) {
   int sv[2];
   grpc_endpoint* ep;
   struct write_socket_state state;
@@ -356,19 +409,27 @@ static void write_test(size_t num_bytes, size_t slice_size) {
       grpc_timespec_to_millis_round_up(grpc_timeout_seconds_to_deadline(20));
   grpc_core::ExecCtx exec_ctx;
 
+  if (collect_timestamps && !grpc_event_engine_can_track_errors()) {
+    return;
+  }
+
   gpr_log(GPR_INFO,
           "Start write test with %" PRIuPTR " bytes, slice size %" PRIuPTR,
           num_bytes, slice_size);
 
-  create_sockets(sv);
+  if (collect_timestamps) {
+    create_inet_sockets(sv);
+  } else {
+    create_sockets(sv);
+  }
 
   grpc_arg a[1];
   a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
   a[0].type = GRPC_ARG_INTEGER,
   a[0].value.integer = static_cast<int>(slice_size);
   grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
-  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test", false), &args,
-                       "test");
+  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test", collect_timestamps),
+                       &args, "test");
   grpc_endpoint_add_to_pollset(ep, g_pollset);
 
   state.ep = ep;
@@ -381,18 +442,26 @@ static void write_test(size_t num_bytes, size_t slice_size) {
   GRPC_CLOSURE_INIT(&write_done_closure, write_done, &state,
                     grpc_schedule_on_exec_ctx);
 
-  grpc_endpoint_write(ep, &outgoing, &write_done_closure);
+  gpr_atm done_timestamps;
+  gpr_atm_rel_store(&done_timestamps, static_cast<gpr_atm>(0));
+  grpc_endpoint_write(ep, &outgoing, &write_done_closure,
+                      grpc_event_engine_can_track_errors() && collect_timestamps
+                          ? (void*)&done_timestamps
+                          : nullptr);
   drain_socket_blocking(sv[0], num_bytes, num_bytes);
+  exec_ctx.Flush();
   gpr_mu_lock(g_mu);
   for (;;) {
     grpc_pollset_worker* worker = nullptr;
-    if (state.write_done) {
+    if (state.write_done &&
+        (!(grpc_event_engine_can_track_errors() && collect_timestamps) ||
+         gpr_atm_acq_load(&done_timestamps) == static_cast<gpr_atm>(1))) {
       break;
     }
     GPR_ASSERT(GRPC_LOG_IF_ERROR(
         "pollset_work", grpc_pollset_work(g_pollset, &worker, deadline)));
     gpr_mu_unlock(g_mu);
-
+    exec_ctx.Flush();
     gpr_mu_lock(g_mu);
   }
   gpr_mu_unlock(g_mu);
@@ -497,14 +566,21 @@ void run_tests(void) {
   large_read_test(8192);
   large_read_test(1);
 
-  write_test(100, 8192);
-  write_test(100, 1);
-  write_test(100000, 8192);
-  write_test(100000, 1);
-  write_test(100000, 137);
+  write_test(100, 8192, false);
+  write_test(100, 1, false);
+  write_test(100000, 8192, false);
+  write_test(100000, 1, false);
+  write_test(100000, 137, false);
+
+  write_test(100, 8192, true);
+  write_test(100, 1, true);
+  write_test(100000, 8192, true);
+  write_test(100000, 1, true);
+  write_test(100, 137, true);
 
   for (i = 1; i < 1000; i = GPR_MAX(i + 1, i * 5 / 4)) {
-    write_test(40320, i);
+    write_test(40320, i, false);
+    write_test(40320, i, true);
   }
 
   release_fd_test(100, 8192);
@@ -549,6 +625,7 @@ int main(int argc, char** argv) {
   grpc_closure destroyed;
   grpc_test_init(argc, argv);
   grpc_init();
+  grpc_core::grpc_tcp_set_write_timestamps_callback(timestamps_verifier);
   {
     grpc_core::ExecCtx exec_ctx;
     g_pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
diff --git a/test/core/iomgr/timer_list_test.cc b/test/core/iomgr/timer_list_test.cc
index b1d919b292..feedf3f149 100644
--- a/test/core/iomgr/timer_list_test.cc
+++ b/test/core/iomgr/timer_list_test.cc
@@ -38,6 +38,8 @@ extern grpc_core::TraceFlag grpc_timer_trace;
 extern grpc_core::TraceFlag grpc_timer_check_trace;
 
 static int cb_called[MAX_CB][2];
+static const int64_t kMillisIn25Days = 2160000000;
+static const int64_t kHoursIn25Days = 600;
 
 static void cb(void* arg, grpc_error* error) {
   cb_called[(intptr_t)arg][error == GRPC_ERROR_NONE]++;
@@ -151,17 +153,112 @@ void destruction_test(void) {
   GPR_ASSERT(1 == cb_called[2][0]);
 }
 
-int main(int argc, char** argv) {
-  grpc_test_init(argc, argv);
-  grpc_core::ExecCtx::GlobalInit();
+/* Cleans up a list with pending timers that simulate long-running-services.
+   This test does the following:
+    1) Simulates grpc server start time to 25 days in the past (completed in
+        `main` using TestOnlyGlobalInit())
+    2) Creates 4 timers - one with a deadline 25 days in the future, one just
+        3 milliseconds in future, one way out in the future, and one using the
+        grpc_timespec_to_millis_round_up function to compute a deadline of 25
+        days in the future
+    3) Simulates 4 milliseconds of elapsed time by changing `now` (cached at
+        step 1) to `now+4`
+    4) Shuts down the timer list
+   https://github.com/grpc/grpc/issues/15904 */
+void long_running_service_cleanup_test(void) {
+  grpc_timer timers[4];
   grpc_core::ExecCtx exec_ctx;
-  grpc_determine_iomgr_platform();
-  grpc_iomgr_platform_init();
-  gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
-  add_test();
-  destruction_test();
-  grpc_iomgr_platform_shutdown();
+
+  gpr_log(GPR_INFO, "long_running_service_cleanup_test");
+
+  grpc_millis now = grpc_core::ExecCtx::Get()->Now();
+  GPR_ASSERT(now >= kMillisIn25Days);
+  grpc_timer_list_init();
+  grpc_core::testing::grpc_tracer_enable_flag(&grpc_timer_trace);
+  grpc_core::testing::grpc_tracer_enable_flag(&grpc_timer_check_trace);
+  memset(cb_called, 0, sizeof(cb_called));
+
+  grpc_timer_init(
+      &timers[0], now + kMillisIn25Days,
+      GRPC_CLOSURE_CREATE(cb, (void*)(intptr_t)0, grpc_schedule_on_exec_ctx));
+  grpc_timer_init(
+      &timers[1], now + 3,
+      GRPC_CLOSURE_CREATE(cb, (void*)(intptr_t)1, grpc_schedule_on_exec_ctx));
+  grpc_timer_init(
+      &timers[2], GRPC_MILLIS_INF_FUTURE - 1,
+      GRPC_CLOSURE_CREATE(cb, (void*)(intptr_t)2, grpc_schedule_on_exec_ctx));
+
+  gpr_timespec deadline_spec = grpc_millis_to_timespec(
+      now + kMillisIn25Days, gpr_clock_type::GPR_CLOCK_MONOTONIC);
+
+  /* grpc_timespec_to_millis_round_up is how users usually compute a millisecond
+    input value into grpc_timer_init, so we mimic that behavior here */
+  grpc_timer_init(
+      &timers[3], grpc_timespec_to_millis_round_up(deadline_spec),
+      GRPC_CLOSURE_CREATE(cb, (void*)(intptr_t)3, grpc_schedule_on_exec_ctx));
+
+  grpc_core::ExecCtx::Get()->TestOnlySetNow(now + 4);
+  GPR_ASSERT(grpc_timer_check(nullptr) == GRPC_TIMERS_FIRED);
+  grpc_core::ExecCtx::Get()->Flush();
+  GPR_ASSERT(0 == cb_called[0][0]);  // Timer 0 not called
+  GPR_ASSERT(0 == cb_called[0][1]);
+  GPR_ASSERT(0 == cb_called[1][0]);
+  GPR_ASSERT(1 == cb_called[1][1]);  // Timer 1 fired
+  GPR_ASSERT(0 == cb_called[2][0]);  // Timer 2 not called
+  GPR_ASSERT(0 == cb_called[2][1]);
+  GPR_ASSERT(0 == cb_called[3][0]);  // Timer 3 not called
+  GPR_ASSERT(0 == cb_called[3][1]);
+
+  grpc_timer_list_shutdown();
+  grpc_core::ExecCtx::Get()->Flush();
+  /* Timers 0, 2, and 3 were fired with an error during cleanup */
+  GPR_ASSERT(1 == cb_called[0][0]);
+  GPR_ASSERT(0 == cb_called[1][0]);
+  GPR_ASSERT(1 == cb_called[2][0]);
+  GPR_ASSERT(1 == cb_called[3][0]);
+}
+
+int main(int argc, char** argv) {
+  /* Tests with default g_start_time */
+  {
+    grpc_test_init(argc, argv);
+    grpc_core::ExecCtx::GlobalInit();
+    grpc_core::ExecCtx exec_ctx;
+    grpc_determine_iomgr_platform();
+    grpc_iomgr_platform_init();
+    gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
+    add_test();
+    destruction_test();
+    grpc_iomgr_platform_shutdown();
+  }
   grpc_core::ExecCtx::GlobalShutdown();
+
+  /* Begin long running service tests */
+  {
+    grpc_test_init(argc, argv);
+    /* Set g_start_time back 25 days. */
+    /* We set g_start_time here in case there are any initialization
+        dependencies that use g_start_time. */
+    gpr_timespec new_start =
+        gpr_time_sub(gpr_now(gpr_clock_type::GPR_CLOCK_MONOTONIC),
+                     gpr_time_from_hours(kHoursIn25Days,
+                                         gpr_clock_type::GPR_CLOCK_MONOTONIC));
+    grpc_core::ExecCtx::TestOnlyGlobalInit(new_start);
+    grpc_core::ExecCtx exec_ctx;
+    grpc_determine_iomgr_platform();
+    grpc_iomgr_platform_init();
+    gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
+#ifndef GPR_WINDOWS
+    /* Skip this test on Windows until we figure out why it fails */
+    /* https://github.com/grpc/grpc/issues/16417 */
+    long_running_service_cleanup_test();
+#endif  // GPR_WINDOWS
+    add_test();
+    destruction_test();
+    grpc_iomgr_platform_shutdown();
+  }
+  grpc_core::ExecCtx::GlobalShutdown();
+
   return 0;
 }
 
diff --git a/test/core/security/credentials_test.cc b/test/core/security/credentials_test.cc
index 8a793e4bb2..97156761bd 100644
--- a/test/core/security/credentials_test.cc
+++ b/test/core/security/credentials_test.cc
@@ -610,7 +610,7 @@ static void test_compute_engine_creds_failure(void) {
   grpc_core::ExecCtx exec_ctx;
   request_metadata_state* state = make_request_metadata_state(
       GRPC_ERROR_CREATE_FROM_STATIC_STRING(
-          "Error occured when fetching oauth2 token."),
+          "Error occurred when fetching oauth2 token."),
       nullptr, 0);
   grpc_auth_metadata_context auth_md_ctx = {test_service_url, test_method,
                                             nullptr, nullptr};
@@ -699,7 +699,7 @@ static void test_refresh_token_creds_failure(void) {
   grpc_core::ExecCtx exec_ctx;
   request_metadata_state* state = make_request_metadata_state(
       GRPC_ERROR_CREATE_FROM_STATIC_STRING(
-          "Error occured when fetching oauth2 token."),
+          "Error occurred when fetching oauth2 token."),
       nullptr, 0);
   grpc_auth_metadata_context auth_md_ctx = {test_service_url, test_method,
                                             nullptr, nullptr};
diff --git a/test/core/surface/completion_queue_test.cc b/test/core/surface/completion_queue_test.cc
index 68129146cc..b889fd0fc6 100644
--- a/test/core/surface/completion_queue_test.cc
+++ b/test/core/surface/completion_queue_test.cc
@@ -22,6 +22,7 @@
 #include <grpc/support/log.h>
 #include <grpc/support/time.h>
 #include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/gprpp/memory.h"
 #include "src/core/lib/iomgr/iomgr.h"
 #include "test/core/util/test_config.h"
 
@@ -41,11 +42,18 @@ static void shutdown_and_destroy(grpc_completion_queue* cc) {
     case GRPC_CQ_NEXT: {
       ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME),
                                       nullptr);
+      GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
       break;
     }
     case GRPC_CQ_PLUCK: {
       ev = grpc_completion_queue_pluck(
           cc, create_test_tag(), gpr_inf_past(GPR_CLOCK_REALTIME), nullptr);
+      GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
+      break;
+    }
+    case GRPC_CQ_CALLBACK: {
+      // Nothing to do here. The shutdown callback will be invoked when
+      // possible.
       break;
     }
     default: {
@@ -54,7 +62,6 @@ static void shutdown_and_destroy(grpc_completion_queue* cc) {
     }
   }
 
-  GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
   grpc_completion_queue_destroy(cc);
 }
 
@@ -350,6 +357,76 @@ static void test_pluck_after_shutdown(void) {
   }
 }
 
+static void test_callback(void) {
+  grpc_completion_queue* cc;
+  void* tags[128];
+  grpc_cq_completion completions[GPR_ARRAY_SIZE(tags)];
+  grpc_cq_polling_type polling_types[] = {
+      GRPC_CQ_DEFAULT_POLLING, GRPC_CQ_NON_LISTENING, GRPC_CQ_NON_POLLING};
+  grpc_completion_queue_attributes attr;
+  unsigned i;
+
+  LOG_TEST("test_callback");
+
+  bool got_shutdown = false;
+  class ShutdownCallback : public grpc_core::CQCallbackInterface {
+   public:
+    ShutdownCallback(bool* done) : done_(done) {}
+    ~ShutdownCallback() {}
+    void Run(bool ok) override { *done_ = ok; }
+
+   private:
+    bool* done_;
+  };
+  ShutdownCallback shutdown_cb(&got_shutdown);
+
+  attr.version = 2;
+  attr.cq_completion_type = GRPC_CQ_CALLBACK;
+  attr.cq_shutdown_cb = &shutdown_cb;
+
+  for (size_t pidx = 0; pidx < GPR_ARRAY_SIZE(polling_types); pidx++) {
+    grpc_core::ExecCtx exec_ctx;  // reset exec_ctx
+    attr.cq_polling_type = polling_types[pidx];
+    cc = grpc_completion_queue_create(
+        grpc_completion_queue_factory_lookup(&attr), &attr, nullptr);
+
+    int counter = 0;
+    class TagCallback : public grpc_core::CQCallbackInterface {
+     public:
+      TagCallback(int* counter, int tag) : counter_(counter), tag_(tag) {}
+      ~TagCallback() {}
+      void Run(bool ok) override {
+        GPR_ASSERT(ok);
+        *counter_ += tag_;
+        grpc_core::Delete(this);
+      };
+
+     private:
+      int* counter_;
+      int tag_;
+    };
+
+    int sumtags = 0;
+    for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
+      tags[i] = static_cast<void*>(grpc_core::New<TagCallback>(&counter, i));
+      sumtags += i;
+    }
+
+    for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
+      GPR_ASSERT(grpc_cq_begin_op(cc, tags[i]));
+      grpc_cq_end_op(cc, tags[i], GRPC_ERROR_NONE, do_nothing_end_completion,
+                     nullptr, &completions[i]);
+    }
+
+    GPR_ASSERT(sumtags == counter);
+
+    shutdown_and_destroy(cc);
+
+    GPR_ASSERT(got_shutdown);
+    got_shutdown = false;
+  }
+}
+
 struct thread_state {
   grpc_completion_queue* cc;
   void* tag;
@@ -368,6 +445,7 @@ int main(int argc, char** argv) {
   test_pluck_after_shutdown();
   test_cq_tls_cache_full();
   test_cq_tls_cache_empty();
+  test_callback();
   grpc_shutdown();
   return 0;
 }
diff --git a/test/core/surface/public_headers_must_be_c89.c b/test/core/surface/public_headers_must_be_c89.c
index 429497df82..a4d2a5bf62 100644
--- a/test/core/surface/public_headers_must_be_c89.c
+++ b/test/core/surface/public_headers_must_be_c89.c
@@ -77,11 +77,13 @@ int main(int argc, char **argv) {
   printf("%lx", (unsigned long) grpc_register_plugin);
   printf("%lx", (unsigned long) grpc_init);
   printf("%lx", (unsigned long) grpc_shutdown);
+  printf("%lx", (unsigned long) grpc_is_initialized);
   printf("%lx", (unsigned long) grpc_version_string);
   printf("%lx", (unsigned long) grpc_g_stands_for);
   printf("%lx", (unsigned long) grpc_completion_queue_factory_lookup);
   printf("%lx", (unsigned long) grpc_completion_queue_create_for_next);
   printf("%lx", (unsigned long) grpc_completion_queue_create_for_pluck);
+  printf("%lx", (unsigned long) grpc_completion_queue_create_for_callback);
   printf("%lx", (unsigned long) grpc_completion_queue_create);
   printf("%lx", (unsigned long) grpc_completion_queue_next);
   printf("%lx", (unsigned long) grpc_completion_queue_pluck);
@@ -131,6 +133,7 @@ int main(int argc, char **argv) {
   printf("%lx", (unsigned long) grpc_resource_quota_ref);
   printf("%lx", (unsigned long) grpc_resource_quota_unref);
   printf("%lx", (unsigned long) grpc_resource_quota_resize);
+  printf("%lx", (unsigned long) grpc_resource_quota_set_max_threads);
   printf("%lx", (unsigned long) grpc_resource_quota_arg_vtable);
   printf("%lx", (unsigned long) grpc_channelz_get_top_channels);
   printf("%lx", (unsigned long) grpc_channelz_get_channel);
diff --git a/test/core/tsi/ssl_transport_security_test.cc b/test/core/tsi/ssl_transport_security_test.cc
index b477904d60..baffad6ea3 100644
--- a/test/core/tsi/ssl_transport_security_test.cc
+++ b/test/core/tsi/ssl_transport_security_test.cc
@@ -208,9 +208,11 @@ static void check_session_reusage(ssl_tsi_test_fixture* ssl_fixture,
       tsi_peer_get_property_by_name(peer, TSI_SSL_SESSION_REUSED_PEER_PROPERTY);
   GPR_ASSERT(session_reused != nullptr);
   if (ssl_fixture->session_reused) {
-    GPR_ASSERT(strcmp(session_reused->value.data, "true") == 0);
+    GPR_ASSERT(strncmp(session_reused->value.data, "true",
+                       session_reused->value.length) == 0);
   } else {
-    GPR_ASSERT(strcmp(session_reused->value.data, "false") == 0);
+    GPR_ASSERT(strncmp(session_reused->value.data, "false",
+                       session_reused->value.length) == 0);
   }
 }
 
diff --git a/test/core/util/mock_endpoint.cc b/test/core/util/mock_endpoint.cc
index 1156cd5fc5..ef6fd62b51 100644
--- a/test/core/util/mock_endpoint.cc
+++ b/test/core/util/mock_endpoint.cc
@@ -55,7 +55,7 @@ static void me_read(grpc_endpoint* ep, grpc_slice_buffer* slices,
 }
 
 static void me_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
-                     grpc_closure* cb) {
+                     grpc_closure* cb, void* arg) {
   mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
   for (size_t i = 0; i < slices->count; i++) {
     m->on_write(slices->slices[i]);
diff --git a/test/core/util/passthru_endpoint.cc b/test/core/util/passthru_endpoint.cc
index 5958216747..3cc8ad6fe1 100644
--- a/test/core/util/passthru_endpoint.cc
+++ b/test/core/util/passthru_endpoint.cc
@@ -76,7 +76,7 @@ static half* other_half(half* h) {
 }
 
 static void me_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
-                     grpc_closure* cb) {
+                     grpc_closure* cb, void* arg) {
   half* m = other_half(reinterpret_cast<half*>(ep));
   gpr_mu_lock(&m->parent->mu);
   grpc_error* error = GRPC_ERROR_NONE;
diff --git a/test/core/util/trickle_endpoint.cc b/test/core/util/trickle_endpoint.cc
index f2efb049b4..62ed72a629 100644
--- a/test/core/util/trickle_endpoint.cc
+++ b/test/core/util/trickle_endpoint.cc
@@ -62,7 +62,7 @@ static void maybe_call_write_cb_locked(trickle_endpoint* te) {
 }
 
 static void te_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
-                     grpc_closure* cb) {
+                     grpc_closure* cb, void* arg) {
   trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
   gpr_mu_lock(&te->mu);
   GPR_ASSERT(te->write_cb == nullptr);
@@ -186,7 +186,8 @@ size_t grpc_trickle_endpoint_trickle(grpc_endpoint* ep) {
       te->last_write = now;
       grpc_endpoint_write(
           te->wrapped, &te->writing_buffer,
-          GRPC_CLOSURE_CREATE(te_finish_write, te, grpc_schedule_on_exec_ctx));
+          GRPC_CLOSURE_CREATE(te_finish_write, te, grpc_schedule_on_exec_ctx),
+          nullptr);
       maybe_call_write_cb_locked(te);
     }
   }