Add unbalanced unary benchmark

1 files changed, 96 insertions, 10 deletions

diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
index fc99b06dbb..b0f7b34bb5 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@@ -77,13 +77,17 @@ static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) {
 
 class TrickledCHTTP2 : public EndpointPairFixture {
  public:
-  TrickledCHTTP2(Service* service, size_t message_size,
+  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",
@@ -242,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;
@@ -314,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 =
@@ -328,8 +329,93 @@ 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()));
+  while (state.KeepRunning()) {
+    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));
+  }
+  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) {
+  for (int i = 1; i <= 128 * 1024 * 1024; i *= 32) {
+    for (int j = 1; j <= 128 * 1024 * 1024; j *= 32) {
+      for (int k = 64; k <= 128 * 1024 * 1024; k *= 16) {
+        double expected_time =
+            static_cast<double>(14 + i + k) / (125.0 * 2 * static_cast<double>(j));
+        if (expected_time > 2.0) continue;
+        b->Args({i, j, k});
+      }
+    }
+  }
+}
+BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
 }
 }