From b6ac9b4573d5f8bc9fbac09482d254b6a7a241b2 Mon Sep 17 00:00:00 2001 From: ncteisen Date: Mon, 8 May 2017 17:10:31 -0700 Subject: [PATCH] Add unbalanced unary benchmark --- .../microbenchmarks/bm_fullstack_trickle.cc | 106 ++++++++++++++++-- 1 file changed, 96 insertions(+), 10 deletions(-) diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc index fc99b06dbb2..b0f7b34bb50 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 fixture( - new TrickledCHTTP2(&service, state.range(0), state.range(1))); + std::unique_ptr 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 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 response_writer; + ServerEnv() : response_writer(&ctx) {} + }; + uint8_t server_env_buffer[2 * sizeof(ServerEnv)]; + ServerEnv* server_env[2] = { + reinterpret_cast(server_env_buffer), + reinterpret_cast(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 stub( + EchoTestService::NewStub(fixture->channel())); + while (state.KeepRunning()) { + GPR_TIMER_SCOPE("BenchmarkCycle", 0); + recv_response.Clear(); + ClientContext cli_ctx; + std::unique_ptr> 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(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(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(14 + i + k) / (125.0 * 2 * static_cast(j)); + if (expected_time > 2.0) continue; + b->Args({i, j, k}); + } + } + } +} +BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs); } }