Merge github.com:grpc/grpc into new_transport_op

13 files changed, 279 insertions, 99 deletions

diff --git a/test/core/channel/channel_stack_test.c b/test/core/channel/channel_stack_test.c
index 76bb57346c..af551c4928 100644
--- a/test/core/channel/channel_stack_test.c
+++ b/test/core/channel/channel_stack_test.c
@@ -68,7 +68,7 @@ static void channel_destroy_func(grpc_exec_ctx *exec_ctx,
 
 static void call_destroy_func(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   ++*(int *)(elem->channel_data);
 }
 
@@ -139,10 +139,16 @@ static void test_create_channel_stack(void) {
   GPR_ASSERT(*channel_data == 0);
 
   call_stack = gpr_malloc(channel_stack->call_stack_size);
-  grpc_error *error =
-      grpc_call_stack_init(&exec_ctx, channel_stack, 1, free_call, call_stack,
-                           NULL, NULL, path, gpr_now(GPR_CLOCK_MONOTONIC),
-                           gpr_inf_future(GPR_CLOCK_MONOTONIC), call_stack);
+  const grpc_call_element_args args = {
+      .call_stack = call_stack,
+      .server_transport_data = NULL,
+      .context = NULL,
+      .path = path,
+      .start_time = gpr_now(GPR_CLOCK_MONOTONIC),
+      .deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC),
+      .arena = NULL};
+  grpc_error *error = grpc_call_stack_init(&exec_ctx, channel_stack, 1,
+                                           free_call, call_stack, &args);
   GPR_ASSERT(error == GRPC_ERROR_NONE);
   GPR_ASSERT(call_stack->count == 1);
   call_elem = grpc_call_stack_element(call_stack, 0);
diff --git a/test/core/end2end/end2end_tests.h b/test/core/end2end/end2end_tests.h
index cb0afd9cd9..cdb26a67e9 100644
--- a/test/core/end2end/end2end_tests.h
+++ b/test/core/end2end/end2end_tests.h
@@ -39,12 +39,15 @@
 typedef struct grpc_end2end_test_fixture grpc_end2end_test_fixture;
 typedef struct grpc_end2end_test_config grpc_end2end_test_config;
 
+/* Test feature flags. */
 #define FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION 1
 #define FEATURE_MASK_SUPPORTS_HOSTNAME_VERIFICATION 2
 #define FEATURE_MASK_SUPPORTS_PER_CALL_CREDENTIALS 4
 #define FEATURE_MASK_SUPPORTS_REQUEST_PROXYING 8
 #define FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL 16
 #define FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER 32
+#define FEATURE_MASK_DOES_NOT_SUPPORT_RESOURCE_QUOTA_SERVER 64
+#define FEATURE_MASK_DOES_NOT_SUPPORT_NETWORK_STATUS_CHANGE 128
 
 #define FAIL_AUTH_CHECK_SERVER_ARG_NAME "fail_auth_check"
 
@@ -56,8 +59,12 @@ struct grpc_end2end_test_fixture {
 };
 
 struct grpc_end2end_test_config {
+  /* A descriptive name for this test fixture. */
   const char *name;
+
+  /* Which features are supported by this fixture. See feature flags above. */
   uint32_t feature_mask;
+
   grpc_end2end_test_fixture (*create_fixture)(grpc_channel_args *client_args,
                                               grpc_channel_args *server_args);
   void (*init_client)(grpc_end2end_test_fixture *f,
diff --git a/test/core/end2end/tests/filter_call_init_fails.c b/test/core/end2end/tests/filter_call_init_fails.c
index d2d6e82d57..65216cf19d 100644
--- a/test/core/end2end/tests/filter_call_init_fails.c
+++ b/test/core/end2end/tests/filter_call_init_fails.c
@@ -213,7 +213,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *and_free_memory) {}
+                              grpc_closure *ignored) {}
 
 static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_channel_element *elem,
diff --git a/test/core/end2end/tests/filter_causes_close.c b/test/core/end2end/tests/filter_causes_close.c
index ebc6f0a52a..6901de04d5 100644
--- a/test/core/end2end/tests/filter_causes_close.c
+++ b/test/core/end2end/tests/filter_causes_close.c
@@ -237,7 +237,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *and_free_memory) {}
+                              grpc_closure *ignored) {}
 
 static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_channel_element *elem,
diff --git a/test/core/end2end/tests/filter_latency.c b/test/core/end2end/tests/filter_latency.c
index d05e9e79a1..2428c92a42 100644
--- a/test/core/end2end/tests/filter_latency.c
+++ b/test/core/end2end/tests/filter_latency.c
@@ -267,7 +267,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_call_element *elem,
                                      const grpc_call_final_info *final_info,
-                                     void *and_free_memory) {
+                                     grpc_closure *ignored) {
   gpr_mu_lock(&g_mu);
   g_client_latency = final_info->stats.latency;
   gpr_mu_unlock(&g_mu);
@@ -276,7 +276,7 @@ static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx,
 static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_call_element *elem,
                                      const grpc_call_final_info *final_info,
-                                     void *and_free_memory) {
+                                     grpc_closure *ignored) {
   gpr_mu_lock(&g_mu);
   g_server_latency = final_info->stats.latency;
   gpr_mu_unlock(&g_mu);
diff --git a/test/core/end2end/tests/network_status_change.c b/test/core/end2end/tests/network_status_change.c
index 7540ce93a1..d7a4106459 100644
--- a/test/core/end2end/tests/network_status_change.c
+++ b/test/core/end2end/tests/network_status_change.c
@@ -240,6 +240,10 @@ static void test_invoke_network_status_change(grpc_end2end_test_config config) {
 }
 
 void network_status_change(grpc_end2end_test_config config) {
+  if (config.feature_mask &
+      FEATURE_MASK_DOES_NOT_SUPPORT_NETWORK_STATUS_CHANGE) {
+    return;
+  }
   test_invoke_network_status_change(config);
 }
 
diff --git a/test/core/end2end/tests/resource_quota_server.c b/test/core/end2end/tests/resource_quota_server.c
index 4f9ed7a3a1..db26b4480e 100644
--- a/test/core/end2end/tests/resource_quota_server.c
+++ b/test/core/end2end/tests/resource_quota_server.c
@@ -113,6 +113,10 @@ static grpc_slice generate_random_slice() {
 }
 
 void resource_quota_server(grpc_end2end_test_config config) {
+  if (config.feature_mask &
+      FEATURE_MASK_DOES_NOT_SUPPORT_RESOURCE_QUOTA_SERVER) {
+    return;
+  }
   grpc_resource_quota *resource_quota =
       grpc_resource_quota_create("test_server");
   grpc_resource_quota_resize(resource_quota, 5 * 1024 * 1024);
diff --git a/test/core/support/cpu_test.c b/test/core/support/cpu_test.c
index ca0fe0ccb5..7b9bf6c5e1 100644
--- a/test/core/support/cpu_test.c
+++ b/test/core/support/cpu_test.c
@@ -81,9 +81,12 @@ static void worker_thread(void *arg) {
   uint32_t cpu;
   unsigned r = 12345678;
   unsigned i, j;
-  for (i = 0; i < 1000 / grpc_test_slowdown_factor(); i++) {
+  /* Avoid repetitive division calculations */
+  int64_t max_i = 1000 / grpc_test_slowdown_factor();
+  int64_t max_j = 1000000 / grpc_test_slowdown_factor();
+  for (i = 0; i < max_i; i++) {
     /* run for a bit - just calculate something random. */
-    for (j = 0; j < 1000000 / grpc_test_slowdown_factor(); j++) {
+    for (j = 0; j < max_j; j++) {
       r = (r * 17) & ((r - i) | (r * i));
     }
     cpu = gpr_cpu_current_cpu();
diff --git a/test/cpp/common/channel_arguments_test.cc b/test/cpp/common/channel_arguments_test.cc
index 190d32ce06..9bcc9f99f6 100644
--- a/test/cpp/common/channel_arguments_test.cc
+++ b/test/cpp/common/channel_arguments_test.cc
@@ -230,13 +230,6 @@ TEST_F(ChannelArgumentsTest, SetSocketMutator) {
   EXPECT_TRUE(HasArg(arg1));
   // arg0 is replaced by arg1
   EXPECT_FALSE(HasArg(arg0));
-
-  // arg0 is destroyed by grpc_socket_mutator_to_arg(mutator1)
-  {
-    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-    arg1.value.pointer.vtable->destroy(&exec_ctx, arg1.value.pointer.p);
-    grpc_exec_ctx_finish(&exec_ctx);
-  }
 }
 
 TEST_F(ChannelArgumentsTest, SetUserAgentPrefix) {
diff --git a/test/cpp/microbenchmarks/bm_call_create.cc b/test/cpp/microbenchmarks/bm_call_create.cc
index 90cd529cf9..fd9033bcef 100644
--- a/test/cpp/microbenchmarks/bm_call_create.cc
+++ b/test/cpp/microbenchmarks/bm_call_create.cc
@@ -232,7 +232,7 @@ static void SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,
 
 static void DestroyCallElem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                             const grpc_call_final_info *final_info,
-                            void *and_free_memory) {}
+                            grpc_closure *then_sched_closure) {}
 
 grpc_error *InitChannelElem(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem,
                             grpc_channel_element_args *args) {
@@ -275,7 +275,7 @@ const char *name;
 /* implementation of grpc_transport_init_stream */
 int InitStream(grpc_exec_ctx *exec_ctx, grpc_transport *self,
                grpc_stream *stream, grpc_stream_refcount *refcount,
-               const void *server_data) {
+               const void *server_data, gpr_arena *arena) {
   return 0;
 }
 
@@ -299,7 +299,7 @@ void PerformOp(grpc_exec_ctx *exec_ctx, grpc_transport *self,
 
 /* implementation of grpc_transport_destroy_stream */
 void DestroyStream(grpc_exec_ctx *exec_ctx, grpc_transport *self,
-                   grpc_stream *stream, void *and_free_memory) {}
+                   grpc_stream *stream, grpc_closure *then_sched_closure) {}
 
 /* implementation of grpc_transport_destroy */
 void Destroy(grpc_exec_ctx *exec_ctx, grpc_transport *self) {}
@@ -397,7 +397,7 @@ static void BM_IsolatedFilter(benchmark::State &state) {
   grpc_channel_stack *channel_stack =
       static_cast<grpc_channel_stack *>(gpr_zalloc(channel_size));
   GPR_ASSERT(GRPC_LOG_IF_ERROR(
-      "call_stack_init",
+      "channel_stack_init",
       grpc_channel_stack_init(&exec_ctx, 1, FilterDestroy, channel_stack,
                               &filters[0], filters.size(), &channel_args,
                               fixture.flags & REQUIRES_TRANSPORT
@@ -412,15 +412,29 @@ static void BM_IsolatedFilter(benchmark::State &state) {
   grpc_slice method = grpc_slice_from_static_string("/foo/bar");
   grpc_call_final_info final_info;
   TestOp test_op_data;
+  grpc_call_element_args call_args;
+  call_args.call_stack = call_stack;
+  call_args.server_transport_data = NULL;
+  call_args.context = NULL;
+  call_args.path = method;
+  call_args.start_time = start_time;
+  call_args.deadline = deadline;
+  const int kArenaSize = 4096;
+  call_args.arena = gpr_arena_create(kArenaSize);
   while (state.KeepRunning()) {
     GRPC_ERROR_UNREF(grpc_call_stack_init(&exec_ctx, channel_stack, 1,
-                                          DoNothing, NULL, NULL, NULL, method,
-                                          start_time, deadline, call_stack));
+                                          DoNothing, NULL, &call_args));
     typename TestOp::Op op(&exec_ctx, &test_op_data, call_stack);
     grpc_call_stack_destroy(&exec_ctx, call_stack, &final_info, NULL);
     op.Finish(&exec_ctx);
     grpc_exec_ctx_flush(&exec_ctx);
+    // recreate arena every 64k iterations to avoid oom
+    if (0 == (state.iterations() & 0xffff)) {
+      gpr_arena_destroy(call_args.arena);
+      call_args.arena = gpr_arena_create(kArenaSize);
+    }
   }
+  gpr_arena_destroy(call_args.arena);
   grpc_channel_stack_destroy(&exec_ctx, channel_stack);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_free(channel_stack);
diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
index dc0e7d769a..513974fde1 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
@@ -54,86 +54,144 @@ auto& force_library_initialization = Library::get();
 
 static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); }
 
-template <class Fixture>
-static void BM_PumpStreamClientToServer(benchmark::State& state) {
+// Repeatedly makes Streaming Bidi calls (exchanging a configurable number of
+// messages in each call) in a loop on a single channel
+//
+//  First parmeter (i.e state.range(0)):  Message size (in bytes) to use
+//  Second parameter (i.e state.range(1)): Number of ping pong messages.
+//      Note: One ping-pong means two messages (one from client to server and
+//      the other from server to client):
+template <class Fixture, class ClientContextMutator, class ServerContextMutator>
+static void BM_StreamingPingPong(benchmark::State& state) {
+  TrackCounters track_counters;
+  const int msg_size = state.range(0);
+  const int max_ping_pongs = state.range(1);
+
   EchoTestService::AsyncService service;
   std::unique_ptr<Fixture> fixture(new Fixture(&service));
   {
+    EchoResponse send_response;
+    EchoResponse recv_response;
     EchoRequest send_request;
     EchoRequest recv_request;
-    if (state.range(0) > 0) {
-      send_request.set_message(std::string(state.range(0), 'a'));
+
+    if (msg_size > 0) {
+      send_request.set_message(std::string(msg_size, 'a'));
+      send_response.set_message(std::string(msg_size, 'b'));
     }
-    Status recv_status;
-    ServerContext svr_ctx;
-    ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
-    service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
-                              fixture->cq(), tag(0));
+
     std::unique_ptr<EchoTestService::Stub> stub(
         EchoTestService::NewStub(fixture->channel()));
-    ClientContext cli_ctx;
-    auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
-    int need_tags = (1 << 0) | (1 << 1);
-    void* t;
-    bool ok;
-    while (need_tags) {
-      GPR_ASSERT(fixture->cq()->Next(&t, &ok));
-      GPR_ASSERT(ok);
-      int i = (int)(intptr_t)t;
-      GPR_ASSERT(need_tags & (1 << i));
-      need_tags &= ~(1 << i);
-    }
-    response_rw.Read(&recv_request, tag(0));
+
     while (state.KeepRunning()) {
-      GPR_TIMER_SCOPE("BenchmarkCycle", 0);
-      request_rw->Write(send_request, tag(1));
-      while (true) {
+      ServerContext svr_ctx;
+      ServerContextMutator svr_ctx_mut(&svr_ctx);
+      ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
+      service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
+                                fixture->cq(), tag(0));
+
+      ClientContext cli_ctx;
+      ClientContextMutator cli_ctx_mut(&cli_ctx);
+      auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
+
+      // Establish async stream between client side and server side
+      void* t;
+      bool ok;
+      int need_tags = (1 << 0) | (1 << 1);
+      while (need_tags) {
         GPR_ASSERT(fixture->cq()->Next(&t, &ok));
-        if (t == tag(0)) {
-          response_rw.Read(&recv_request, tag(0));
-        } else if (t == tag(1)) {
-          break;
-        } else {
-          GPR_ASSERT(false);
+        GPR_ASSERT(ok);
+        int i = (int)(intptr_t)t;
+        GPR_ASSERT(need_tags & (1 << i));
+        need_tags &= ~(1 << i);
+      }
+
+      // Send 'max_ping_pongs' number of ping pong messages
+      int ping_pong_cnt = 0;
+      while (ping_pong_cnt < max_ping_pongs) {
+        request_rw->Write(send_request, tag(0));   // Start client send
+        response_rw.Read(&recv_request, tag(1));   // Start server recv
+        request_rw->Read(&recv_response, tag(2));  // Start client recv
+
+        need_tags = (1 << 0) | (1 << 1) | (1 << 2) | (1 << 3);
+        while (need_tags) {
+          GPR_ASSERT(fixture->cq()->Next(&t, &ok));
+          GPR_ASSERT(ok);
+          int i = (int)(intptr_t)t;
+
+          // If server recv is complete, start the server send operation
+          if (i == 1) {
+            response_rw.Write(send_response, tag(3));
+          }
+
+          GPR_ASSERT(need_tags & (1 << i));
+          need_tags &= ~(1 << i);
         }
+
+        ping_pong_cnt++;
       }
-    }
-    request_rw->WritesDone(tag(1));
-    need_tags = (1 << 0) | (1 << 1);
-    while (need_tags) {
-      GPR_ASSERT(fixture->cq()->Next(&t, &ok));
-      int i = (int)(intptr_t)t;
-      GPR_ASSERT(need_tags & (1 << i));
-      need_tags &= ~(1 << i);
+
+      request_rw->WritesDone(tag(0));
+      response_rw.Finish(Status::OK, tag(1));
+
+      Status recv_status;
+      request_rw->Finish(&recv_status, tag(2));
+
+      need_tags = (1 << 0) | (1 << 1) | (1 << 2);
+      while (need_tags) {
+        GPR_ASSERT(fixture->cq()->Next(&t, &ok));
+        int i = (int)(intptr_t)t;
+        GPR_ASSERT(need_tags & (1 << i));
+        need_tags &= ~(1 << i);
+      }
+
+      GPR_ASSERT(recv_status.ok());
     }
   }
+
   fixture->Finish(state);
   fixture.reset();
-  state.SetBytesProcessed(state.range(0) * state.iterations());
+  state.SetBytesProcessed(msg_size * state.iterations() * max_ping_pongs * 2);
+  track_counters.Finish(state);
 }
 
-template <class Fixture>
-static void BM_PumpStreamServerToClient(benchmark::State& state) {
+// Repeatedly sends ping pong messages in a single streaming Bidi call in a loop
+//     First parmeter (i.e state.range(0)):  Message size (in bytes) to use
+template <class Fixture, class ClientContextMutator, class ServerContextMutator>
+static void BM_StreamingPingPongMsgs(benchmark::State& state) {
+  TrackCounters track_counters;
+  const int msg_size = state.range(0);
+
   EchoTestService::AsyncService service;
   std::unique_ptr<Fixture> fixture(new Fixture(&service));
   {
     EchoResponse send_response;
     EchoResponse recv_response;
-    if (state.range(0) > 0) {
-      send_response.set_message(std::string(state.range(0), 'a'));
+    EchoRequest send_request;
+    EchoRequest recv_request;
+
+    if (msg_size > 0) {
+      send_request.set_message(std::string(msg_size, 'a'));
+      send_response.set_message(std::string(msg_size, 'b'));
     }
-    Status recv_status;
+
+    std::unique_ptr<EchoTestService::Stub> stub(
+        EchoTestService::NewStub(fixture->channel()));
+
     ServerContext svr_ctx;
+    ServerContextMutator svr_ctx_mut(&svr_ctx);
     ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
     service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
                               fixture->cq(), tag(0));
-    std::unique_ptr<EchoTestService::Stub> stub(
-        EchoTestService::NewStub(fixture->channel()));
+
     ClientContext cli_ctx;
+    ClientContextMutator cli_ctx_mut(&cli_ctx);
     auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
-    int need_tags = (1 << 0) | (1 << 1);
+
+    // Establish async stream between client side and server side
     void* t;
     bool ok;
+    int need_tags = (1 << 0) | (1 << 1);
     while (need_tags) {
       GPR_ASSERT(fixture->cq()->Next(&t, &ok));
       GPR_ASSERT(ok);
@@ -141,54 +199,79 @@ static void BM_PumpStreamServerToClient(benchmark::State& state) {
       GPR_ASSERT(need_tags & (1 << i));
       need_tags &= ~(1 << i);
     }
-    request_rw->Read(&recv_response, tag(0));
+
     while (state.KeepRunning()) {
       GPR_TIMER_SCOPE("BenchmarkCycle", 0);
-      response_rw.Write(send_response, tag(1));
-      while (true) {
+      request_rw->Write(send_request, tag(0));   // Start client send
+      response_rw.Read(&recv_request, tag(1));   // Start server recv
+      request_rw->Read(&recv_response, tag(2));  // Start client recv
+
+      need_tags = (1 << 0) | (1 << 1) | (1 << 2) | (1 << 3);
+      while (need_tags) {
         GPR_ASSERT(fixture->cq()->Next(&t, &ok));
-        if (t == tag(0)) {
-          request_rw->Read(&recv_response, tag(0));
-        } else if (t == tag(1)) {
-          break;
-        } else {
-          GPR_ASSERT(false);
+        GPR_ASSERT(ok);
+        int i = (int)(intptr_t)t;
+
+        // If server recv is complete, start the server send operation
+        if (i == 1) {
+          response_rw.Write(send_response, tag(3));
         }
+
+        GPR_ASSERT(need_tags & (1 << i));
+        need_tags &= ~(1 << i);
       }
     }
+
+    request_rw->WritesDone(tag(0));
     response_rw.Finish(Status::OK, tag(1));
-    need_tags = (1 << 0) | (1 << 1);
+    Status recv_status;
+    request_rw->Finish(&recv_status, tag(2));
+
+    need_tags = (1 << 0) | (1 << 1) | (1 << 2);
     while (need_tags) {
       GPR_ASSERT(fixture->cq()->Next(&t, &ok));
       int i = (int)(intptr_t)t;
       GPR_ASSERT(need_tags & (1 << i));
       need_tags &= ~(1 << i);
     }
+
+    GPR_ASSERT(recv_status.ok());
   }
+
   fixture->Finish(state);
   fixture.reset();
-  state.SetBytesProcessed(state.range(0) * state.iterations());
+  state.SetBytesProcessed(msg_size * state.iterations() * 2);
+  track_counters.Finish(state);
 }
 
 /*******************************************************************************
  * CONFIGURATIONS
  */
 
-BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, TCP)
-    ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, UDS)
-    ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, SockPair)
-    ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, InProcessCHTTP2)
-    ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, TCP)
-    ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, UDS)
-    ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair)
+// Generate Args for StreamingPingPong benchmarks. Currently generates args for
+// only "small streams" (i.e streams with 0, 1 or 2 messages)
+static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) {
+  int msg_size = 0;
+
+  b->Args({0, 0});  // spl case: 0 ping-pong msgs (msg_size doesn't matter here)
+
+  for (msg_size = 0; msg_size <= 128 * 1024 * 1024;
+       msg_size == 0 ? msg_size++ : msg_size *= 8) {
+    b->Args({msg_size, 1});
+    b->Args({msg_size, 2});
+  }
+}
+
+BENCHMARK_TEMPLATE(BM_StreamingPingPong, InProcessCHTTP2, NoOpMutator,
+                   NoOpMutator)
+    ->Apply(StreamingPingPongArgs);
+BENCHMARK_TEMPLATE(BM_StreamingPingPong, TCP, NoOpMutator, NoOpMutator)
+    ->Apply(StreamingPingPongArgs);
+
+BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcessCHTTP2, NoOpMutator,
+                   NoOpMutator)
     ->Range(0, 128 * 1024 * 1024);
-BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2)
+BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, TCP, NoOpMutator, NoOpMutator)
     ->Range(0, 128 * 1024 * 1024);
 
 }  // namespace testing
diff --git a/test/http2_test/http2_server_health_check.py b/test/http2_test/http2_server_health_check.py
new file mode 100644
index 0000000000..dd9402b855
--- /dev/null
+++ b/test/http2_test/http2_server_health_check.py
@@ -0,0 +1,49 @@
+# Copyright 2017, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import argparse
+import hyper
+import sys
+
+# Utility to healthcheck the http2 server. Used when starting the server to
+# verify that the server is live before tests begin.
+if __name__ == '__main__':
+  parser = argparse.ArgumentParser()
+  parser.add_argument('--server_host', type=str, default='localhost')
+  parser.add_argument('--server_port', type=int, default=8080)
+  args = parser.parse_args()
+  server_host = args.server_host
+  server_port = args.server_port
+  conn = hyper.HTTP20Connection('%s:%d' % (server_host, server_port))
+  conn.request('POST', '/grpc.testing.TestService/UnaryCall')
+  resp = conn.get_response()
+  if resp.headers.get('grpc-encoding') is None:
+    sys.exit(1)
+  else:
+    sys.exit(0)
diff --git a/test/http2_test/http2_test_server.py b/test/http2_test/http2_test_server.py
index abde3433ad..46c3e00d18 100644
--- a/test/http2_test/http2_test_server.py
+++ b/test/http2_test/http2_test_server.py
@@ -31,6 +31,7 @@
 
 import argparse
 import logging
+import sys
 import twisted
 import twisted.internet
 import twisted.internet.endpoints
@@ -53,9 +54,11 @@ _TEST_CASE_MAPPING = {
   'max_streams': test_max_streams.TestcaseSettingsMaxStreams,
 }
 
+_exit_code = 0
+
 class H2Factory(twisted.internet.protocol.Factory):
   def __init__(self, testcase):
-    logging.info('Creating H2Factory for new connection.')
+    logging.info('Creating H2Factory for new connection (%s)', testcase)
     self._num_streams = 0
     self._testcase = testcase
 
@@ -83,6 +86,17 @@ def parse_arguments():
     )
   return parser.parse_args()
 
+def listen(endpoint, test_case):
+  deferred = endpoint.listen(H2Factory(test_case))
+  def listen_error(reason):
+    # If listening fails, we stop the reactor and exit the program
+    # with exit code 1.
+    global _exit_code
+    _exit_code = 1
+    logging.error('Listening failed: %s' % reason.value)
+    twisted.internet.reactor.stop()
+  deferred.addErrback(listen_error)
+
 def start_test_servers(base_port):
   """ Start one server per test case on incrementing port numbers
   beginning with base_port """
@@ -92,7 +106,9 @@ def start_test_servers(base_port):
     logging.warning('serving on port %d : %s'%(portnum, test_case))
     endpoint = twisted.internet.endpoints.TCP4ServerEndpoint(
       twisted.internet.reactor, portnum, backlog=128)
-    endpoint.listen(H2Factory(test_case))
+    # Wait until the reactor is running before calling endpoint.listen().
+    twisted.internet.reactor.callWhenRunning(listen, endpoint, test_case)
+
     index += 1
 
 if __name__ == '__main__':
@@ -102,3 +118,4 @@ if __name__ == '__main__':
   args = parse_arguments()
   start_test_servers(args.base_port)
   twisted.internet.reactor.run()
+  sys.exit(_exit_code)