Merge branch 'master' of https://github.com/grpc/grpc into frame-size

7 files changed, 145 insertions, 39 deletions

diff --git a/test/core/end2end/cq_verifier.c b/test/core/end2end/cq_verifier.c
index 5eea5d43fe..0fafb0c8c9 100644
--- a/test/core/end2end/cq_verifier.c
+++ b/test/core/end2end/cq_verifier.c
@@ -77,7 +77,7 @@ struct cq_verifier {
 };
 
 cq_verifier *cq_verifier_create(grpc_completion_queue *cq) {
-  cq_verifier *v = gpr_malloc(sizeof(cq_verifier));
+  cq_verifier *v = (cq_verifier *)gpr_malloc(sizeof(cq_verifier));
   v->cq = cq;
   v->first_expectation = NULL;
   return v;
@@ -314,7 +314,7 @@ void cq_verify_empty(cq_verifier *v) { cq_verify_empty_timeout(v, 1); }
 
 static void add(cq_verifier *v, const char *file, int line,
                 grpc_completion_type type, void *tag, bool success) {
-  expectation *e = gpr_malloc(sizeof(expectation));
+  expectation *e = (expectation *)gpr_malloc(sizeof(expectation));
   e->type = type;
   e->file = file;
   e->line = line;
diff --git a/test/core/end2end/fake_resolver.c b/test/core/end2end/fake_resolver.c
index df902a24bf..736b224fd6 100644
--- a/test/core/end2end/fake_resolver.c
+++ b/test/core/end2end/fake_resolver.c
@@ -136,7 +136,7 @@ struct grpc_fake_resolver_response_generator {
 grpc_fake_resolver_response_generator*
 grpc_fake_resolver_response_generator_create() {
   grpc_fake_resolver_response_generator* generator =
-      gpr_zalloc(sizeof(*generator));
+      (grpc_fake_resolver_response_generator*)gpr_zalloc(sizeof(*generator));
   gpr_ref_init(&generator->refcount, 1);
   return generator;
 }
@@ -157,7 +157,8 @@ void grpc_fake_resolver_response_generator_unref(
 
 static void set_response_cb(grpc_exec_ctx* exec_ctx, void* arg,
                             grpc_error* error) {
-  grpc_fake_resolver_response_generator* generator = arg;
+  grpc_fake_resolver_response_generator* generator =
+      (grpc_fake_resolver_response_generator*)arg;
   fake_resolver* r = generator->resolver;
   if (r->next_results != NULL) {
     grpc_channel_args_destroy(exec_ctx, r->next_results);
@@ -180,11 +181,13 @@ void grpc_fake_resolver_response_generator_set_response(
 }
 
 static void* response_generator_arg_copy(void* p) {
-  return grpc_fake_resolver_response_generator_ref(p);
+  return grpc_fake_resolver_response_generator_ref(
+      (grpc_fake_resolver_response_generator*)p);
 }
 
 static void response_generator_arg_destroy(grpc_exec_ctx* exec_ctx, void* p) {
-  grpc_fake_resolver_response_generator_unref(p);
+  grpc_fake_resolver_response_generator_unref(
+      (grpc_fake_resolver_response_generator*)p);
 }
 
 static int response_generator_cmp(void* a, void* b) { return GPR_ICMP(a, b); }
@@ -208,7 +211,7 @@ grpc_fake_resolver_get_response_generator(const grpc_channel_args* args) {
   const grpc_arg* arg =
       grpc_channel_args_find(args, GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR);
   if (arg == NULL || arg->type != GRPC_ARG_POINTER) return NULL;
-  return arg->value.pointer.p;
+  return (grpc_fake_resolver_response_generator*)arg->value.pointer.p;
 }
 
 //
@@ -222,7 +225,7 @@ static void fake_resolver_factory_unref(grpc_resolver_factory* factory) {}
 static grpc_resolver* fake_resolver_create(grpc_exec_ctx* exec_ctx,
                                            grpc_resolver_factory* factory,
                                            grpc_resolver_args* args) {
-  fake_resolver* r = gpr_zalloc(sizeof(*r));
+  fake_resolver* r = (fake_resolver*)gpr_zalloc(sizeof(*r));
   r->channel_args = grpc_channel_args_copy(args->args);
   grpc_resolver_init(&r->base, &fake_resolver_vtable, args->combiner);
   grpc_fake_resolver_response_generator* response_generator =
diff --git a/test/core/end2end/fixtures/http_proxy_fixture.c b/test/core/end2end/fixtures/http_proxy_fixture.c
index f0d09487c6..5df43b9384 100644
--- a/test/core/end2end/fixtures/http_proxy_fixture.c
+++ b/test/core/end2end/fixtures/http_proxy_fixture.c
@@ -156,7 +156,7 @@ static void proxy_connection_failed(grpc_exec_ctx* exec_ctx,
 // Callback for writing proxy data to the client.
 static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
                                  grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   if (error != GRPC_ERROR_NONE) {
     proxy_connection_failed(exec_ctx, conn, true /* is_client */,
                             "HTTP proxy client write", error);
@@ -181,7 +181,7 @@ static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
 // Callback for writing proxy data to the backend server.
 static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
                                  grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   if (error != GRPC_ERROR_NONE) {
     proxy_connection_failed(exec_ctx, conn, false /* is_client */,
                             "HTTP proxy server write", error);
@@ -207,7 +207,7 @@ static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
 // the backend server.
 static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
                                 grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   if (error != GRPC_ERROR_NONE) {
     proxy_connection_failed(exec_ctx, conn, true /* is_client */,
                             "HTTP proxy client read", error);
@@ -239,7 +239,7 @@ static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
 // proxied to the client.
 static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
                                 grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   if (error != GRPC_ERROR_NONE) {
     proxy_connection_failed(exec_ctx, conn, false /* is_client */,
                             "HTTP proxy server read", error);
@@ -270,7 +270,7 @@ static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
 // Callback to write the HTTP response for the CONNECT request.
 static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg,
                                    grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   if (error != GRPC_ERROR_NONE) {
     proxy_connection_failed(exec_ctx, conn, true /* is_client */,
                             "HTTP proxy write response", error);
@@ -294,7 +294,7 @@ static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg,
 // CONNECT request.
 static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
                                    grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   if (error != GRPC_ERROR_NONE) {
     // TODO(roth): Technically, in this case, we should handle the error
     // by returning an HTTP response to the client indicating that the
@@ -324,7 +324,7 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
 // which will cause the client connection to be dropped.
 static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
                                  grpc_error* error) {
-  proxy_connection* conn = arg;
+  proxy_connection* conn = (proxy_connection*)arg;
   gpr_log(GPR_DEBUG, "on_read_request_done: %p %s", conn,
           grpc_error_string(error));
   if (error != GRPC_ERROR_NONE) {
@@ -389,9 +389,9 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg,
                       grpc_endpoint* endpoint, grpc_pollset* accepting_pollset,
                       grpc_tcp_server_acceptor* acceptor) {
   gpr_free(acceptor);
-  grpc_end2end_http_proxy* proxy = arg;
+  grpc_end2end_http_proxy* proxy = (grpc_end2end_http_proxy*)arg;
   // Instantiate proxy_connection.
-  proxy_connection* conn = gpr_zalloc(sizeof(*conn));
+  proxy_connection* conn = (proxy_connection*)gpr_zalloc(sizeof(*conn));
   gpr_ref(&proxy->users);
   conn->client_endpoint = endpoint;
   conn->proxy = proxy;
@@ -430,7 +430,7 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg,
 //
 
 static void thread_main(void* arg) {
-  grpc_end2end_http_proxy* proxy = arg;
+  grpc_end2end_http_proxy* proxy = (grpc_end2end_http_proxy*)arg;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   do {
     gpr_ref(&proxy->users);
@@ -450,7 +450,8 @@ static void thread_main(void* arg) {
 
 grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) {
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  grpc_end2end_http_proxy* proxy = gpr_malloc(sizeof(*proxy));
+  grpc_end2end_http_proxy* proxy =
+      (grpc_end2end_http_proxy*)gpr_malloc(sizeof(*proxy));
   memset(proxy, 0, sizeof(*proxy));
   gpr_ref_init(&proxy->users, 1);
   // Construct proxy address.
@@ -473,7 +474,7 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) {
   GPR_ASSERT(error == GRPC_ERROR_NONE);
   GPR_ASSERT(port == proxy_port);
   // Start server.
-  proxy->pollset = gpr_zalloc(grpc_pollset_size());
+  proxy->pollset = (grpc_pollset*)gpr_zalloc(grpc_pollset_size());
   grpc_pollset_init(proxy->pollset, &proxy->mu);
   grpc_tcp_server_start(&exec_ctx, proxy->server, &proxy->pollset, 1, on_accept,
                         proxy);
@@ -487,7 +488,7 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) {
 
 static void destroy_pollset(grpc_exec_ctx* exec_ctx, void* arg,
                             grpc_error* error) {
-  grpc_pollset* pollset = arg;
+  grpc_pollset* pollset = (grpc_pollset*)arg;
   grpc_pollset_destroy(pollset);
   gpr_free(pollset);
 }
diff --git a/test/core/security/oauth2_utils.c b/test/core/security/oauth2_utils.c
index f0550db1d0..838625705d 100644
--- a/test/core/security/oauth2_utils.c
+++ b/test/core/security/oauth2_utils.c
@@ -55,7 +55,7 @@ static void on_oauth2_response(grpc_exec_ctx *exec_ctx, void *user_data,
                                grpc_credentials_md *md_elems, size_t num_md,
                                grpc_credentials_status status,
                                const char *error_details) {
-  oauth2_request *request = user_data;
+  oauth2_request *request = (oauth2_request *)user_data;
   char *token = NULL;
   grpc_slice token_slice;
   if (status == GRPC_CREDENTIALS_ERROR) {
@@ -63,7 +63,7 @@ static void on_oauth2_response(grpc_exec_ctx *exec_ctx, void *user_data,
   } else {
     GPR_ASSERT(num_md == 1);
     token_slice = md_elems[0].value;
-    token = gpr_malloc(GRPC_SLICE_LENGTH(token_slice) + 1);
+    token = (char *)gpr_malloc(GRPC_SLICE_LENGTH(token_slice) + 1);
     memcpy(token, GRPC_SLICE_START_PTR(token_slice),
            GRPC_SLICE_LENGTH(token_slice));
     token[GRPC_SLICE_LENGTH(token_slice)] = '\0';
@@ -87,7 +87,7 @@ char *grpc_test_fetch_oauth2_token_with_credentials(
   grpc_closure do_nothing_closure;
   grpc_auth_metadata_context null_ctx = {"", "", NULL, NULL};
 
-  grpc_pollset *pollset = gpr_zalloc(grpc_pollset_size());
+  grpc_pollset *pollset = (grpc_pollset *)gpr_zalloc(grpc_pollset_size());
   grpc_pollset_init(pollset, &request.mu);
   request.pops = grpc_polling_entity_create_from_pollset(pollset);
   request.is_done = 0;
diff --git a/test/cpp/microbenchmarks/bm_call_create.cc b/test/cpp/microbenchmarks/bm_call_create.cc
index c91219e98c..67e7c02535 100644
--- a/test/cpp/microbenchmarks/bm_call_create.cc
+++ b/test/cpp/microbenchmarks/bm_call_create.cc
@@ -563,7 +563,8 @@ static void BM_IsolatedFilter(benchmark::State &state) {
   }
 
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  size_t channel_size = grpc_channel_stack_size(&filters[0], filters.size());
+  size_t channel_size = grpc_channel_stack_size(
+      filters.size() == 0 ? NULL : &filters[0], filters.size());
   grpc_channel_stack *channel_stack =
       static_cast<grpc_channel_stack *>(gpr_zalloc(channel_size));
   GPR_ASSERT(GRPC_LOG_IF_ERROR(
diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
index 01d2023fa6..d7e3a9cf47 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@@ -53,7 +53,7 @@ DEFINE_int32(
     "Number of megabytes to pump before collecting flow control stats");
 DEFINE_int32(
     warmup_iterations, 100,
-    "Number of megabytes to pump before collecting flow control stats");
+    "Number of iterations to run before collecting flow control stats");
 DEFINE_int32(warmup_max_time_seconds, 10,
              "Maximum number of seconds to run warmup loop");
 
@@ -77,13 +77,14 @@ static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) {
 
 class TrickledCHTTP2 : public EndpointPairFixture {
  public:
-  TrickledCHTTP2(Service* service, size_t message_size,
-                 size_t kilobits_per_second)
+  TrickledCHTTP2(Service* service, bool streaming, size_t req_size,
+                 size_t resp_size, size_t kilobits_per_second)
       : EndpointPairFixture(service, MakeEndpoints(kilobits_per_second),
                             FixtureConfiguration()) {
     if (FLAGS_log) {
       std::ostringstream fn;
-      fn << "trickle." << message_size << "." << kilobits_per_second << ".csv";
+      fn << "trickle." << (streaming ? "streaming" : "unary") << "." << req_size
+         << "." << resp_size << "." << kilobits_per_second << ".csv";
       log_.reset(new std::ofstream(fn.str().c_str()));
       write_csv(log_.get(), "t", "iteration", "client_backlog",
                 "server_backlog", "client_t_stall", "client_s_stall",
@@ -245,8 +246,9 @@ static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok,
 
 static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
   EchoTestService::AsyncService service;
-  std::unique_ptr<TrickledCHTTP2> fixture(
-      new TrickledCHTTP2(&service, state.range(0), state.range(1)));
+  std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
+      &service, true, state.range(0) /* req_size */,
+      state.range(0) /* resp_size */, state.range(1) /* bw in kbit/s */));
   {
     EchoResponse send_response;
     EchoResponse recv_response;
@@ -317,11 +319,7 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
   state.SetBytesProcessed(state.range(0) * state.iterations());
 }
 
-/*******************************************************************************
- * CONFIGURATIONS
- */
-
-static void TrickleArgs(benchmark::internal::Benchmark* b) {
+static void StreamingTrickleArgs(benchmark::internal::Benchmark* b) {
   for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
     for (int j = 64; j <= 128 * 1024 * 1024; j *= 8) {
       double expected_time =
@@ -331,8 +329,111 @@ static void TrickleArgs(benchmark::internal::Benchmark* b) {
     }
   }
 }
+BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(StreamingTrickleArgs);
 
-BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs);
+static void BM_PumpUnbalancedUnary_Trickle(benchmark::State& state) {
+  EchoTestService::AsyncService service;
+  std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
+      &service, true, state.range(0) /* req_size */,
+      state.range(1) /* resp_size */, state.range(2) /* bw in kbit/s */));
+  EchoRequest send_request;
+  EchoResponse send_response;
+  EchoResponse recv_response;
+  if (state.range(0) > 0) {
+    send_request.set_message(std::string(state.range(0), 'a'));
+  }
+  if (state.range(1) > 0) {
+    send_response.set_message(std::string(state.range(1), 'a'));
+  }
+  Status recv_status;
+  struct ServerEnv {
+    ServerContext ctx;
+    EchoRequest recv_request;
+    grpc::ServerAsyncResponseWriter<EchoResponse> response_writer;
+    ServerEnv() : response_writer(&ctx) {}
+  };
+  uint8_t server_env_buffer[2 * sizeof(ServerEnv)];
+  ServerEnv* server_env[2] = {
+      reinterpret_cast<ServerEnv*>(server_env_buffer),
+      reinterpret_cast<ServerEnv*>(server_env_buffer + sizeof(ServerEnv))};
+  new (server_env[0]) ServerEnv;
+  new (server_env[1]) ServerEnv;
+  service.RequestEcho(&server_env[0]->ctx, &server_env[0]->recv_request,
+                      &server_env[0]->response_writer, fixture->cq(),
+                      fixture->cq(), tag(0));
+  service.RequestEcho(&server_env[1]->ctx, &server_env[1]->recv_request,
+                      &server_env[1]->response_writer, fixture->cq(),
+                      fixture->cq(), tag(1));
+  std::unique_ptr<EchoTestService::Stub> stub(
+      EchoTestService::NewStub(fixture->channel()));
+  auto inner_loop = [&](bool in_warmup) {
+    GPR_TIMER_SCOPE("BenchmarkCycle", 0);
+    recv_response.Clear();
+    ClientContext cli_ctx;
+    std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
+        stub->AsyncEcho(&cli_ctx, send_request, fixture->cq()));
+    void* t;
+    bool ok;
+    TrickleCQNext(fixture.get(), &t, &ok, state.iterations());
+    GPR_ASSERT(ok);
+    GPR_ASSERT(t == tag(0) || t == tag(1));
+    intptr_t slot = reinterpret_cast<intptr_t>(t);
+    ServerEnv* senv = server_env[slot];
+    senv->response_writer.Finish(send_response, Status::OK, tag(3));
+    response_reader->Finish(&recv_response, &recv_status, tag(4));
+    for (int i = (1 << 3) | (1 << 4); i != 0;) {
+      TrickleCQNext(fixture.get(), &t, &ok, state.iterations());
+      GPR_ASSERT(ok);
+      int tagnum = (int)reinterpret_cast<intptr_t>(t);
+      GPR_ASSERT(i & (1 << tagnum));
+      i -= 1 << tagnum;
+    }
+    GPR_ASSERT(recv_status.ok());
+
+    senv->~ServerEnv();
+    senv = new (senv) ServerEnv();
+    service.RequestEcho(&senv->ctx, &senv->recv_request, &senv->response_writer,
+                        fixture->cq(), fixture->cq(), tag(slot));
+  };
+  gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
+  for (int i = 0;
+       i < GPR_MAX(FLAGS_warmup_iterations, FLAGS_warmup_megabytes * 1024 *
+                                                1024 / (14 + state.range(0)));
+       i++) {
+    inner_loop(true);
+    if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), warmup_start),
+                     gpr_time_from_seconds(FLAGS_warmup_max_time_seconds,
+                                           GPR_TIMESPAN)) > 0) {
+      break;
+    }
+  }
+  while (state.KeepRunning()) {
+    inner_loop(false);
+  }
+  fixture->Finish(state);
+  fixture.reset();
+  server_env[0]->~ServerEnv();
+  server_env[1]->~ServerEnv();
+  state.SetBytesProcessed(state.range(0) * state.iterations() +
+                          state.range(1) * state.iterations());
+}
+
+static void UnaryTrickleArgs(benchmark::internal::Benchmark* b) {
+  const int cli_1024k = 1024 * 1024;
+  const int cli_32M = 32 * 1024 * 1024;
+  const int svr_256k = 256 * 1024;
+  const int svr_4M = 4 * 1024 * 1024;
+  const int svr_64M = 64 * 1024 * 1024;
+  for (int bw = 64; bw <= 128 * 1024 * 1024; bw *= 16) {
+    b->Args({bw, cli_1024k, svr_256k});
+    b->Args({bw, cli_1024k, svr_4M});
+    b->Args({bw, cli_1024k, svr_64M});
+    b->Args({bw, cli_32M, svr_256k});
+    b->Args({bw, cli_32M, svr_4M});
+    b->Args({bw, cli_32M, svr_64M});
+  }
+}
+BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
 }
 }
 
diff --git a/test/cpp/performance/writes_per_rpc_test.cc b/test/cpp/performance/writes_per_rpc_test.cc
index 7a914c1547..12d8268330 100644
--- a/test/cpp/performance/writes_per_rpc_test.cc
+++ b/test/cpp/performance/writes_per_rpc_test.cc
@@ -254,8 +254,8 @@ TEST(WritesPerRpcTest, UnaryPingPong) {
   EXPECT_LT(UnaryPingPong(0, 0), 2.05);
   EXPECT_LT(UnaryPingPong(1, 0), 2.05);
   EXPECT_LT(UnaryPingPong(0, 1), 2.05);
-  EXPECT_LT(UnaryPingPong(4096, 0), 2.2);
-  EXPECT_LT(UnaryPingPong(0, 4096), 2.2);
+  EXPECT_LT(UnaryPingPong(4096, 0), 2.5);
+  EXPECT_LT(UnaryPingPong(0, 4096), 2.5);
 }
 
 }  // namespace testing