From e3ffd861c949e1841065784ae2dadb4348b19ab2 Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Fri, 3 Feb 2017 19:39:31 -0800 Subject: [PATCH 1/4] New microbenchmarks for Streaming Ping pong calls and messages --- test/cpp/microbenchmarks/bm_fullstack.cc | 207 +++++++++++++++++++++++ 1 file changed, 207 insertions(+) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index c3e96c572c8..611ee8ddb3a 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -439,6 +439,194 @@ static void BM_UnaryPingPong(benchmark::State& state) { state.range(1) * state.iterations()); } +// 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 +static void BM_StreamingPingPong(benchmark::State& state) { + const int msg_size = state.range(0); + const int max_ping_pongs = state.range(1); + + EchoTestService::AsyncService service; + std::unique_ptr fixture(new Fixture(&service)); + { + EchoResponse send_response; + EchoResponse recv_response; + 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')); + } + + std::unique_ptr stub( + EchoTestService::NewStub(fixture->channel())); + + while (state.KeepRunning()) { + ServerContext svr_ctx; + ServerContextMutator svr_ctx_mut(&svr_ctx); + ServerAsyncReaderWriter 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)); + 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(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(msg_size * state.iterations() * max_ping_pongs * 2); +} + +// 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 +static void BM_StreamingPingPongMsgs(benchmark::State& state) { + const int msg_size = state.range(0); + int ping_pong_cnt = 0; + + EchoTestService::AsyncService service; + std::unique_ptr fixture(new Fixture(&service)); + { + EchoResponse send_response; + EchoResponse recv_response; + 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')); + } + + std::unique_ptr stub( + EchoTestService::NewStub(fixture->channel())); + + ServerContext svr_ctx; + ServerContextMutator svr_ctx_mut(&svr_ctx); + ServerAsyncReaderWriter 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)); + GPR_ASSERT(ok); + int i = (int)(intptr_t)t; + GPR_ASSERT(need_tags & (1 << i)); + need_tags &= ~(1 << i); + } + + while (state.KeepRunning()) { + 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(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(msg_size * state.iterations() * ping_pong_cnt * 2); +} + template static void BM_PumpStreamClientToServer(benchmark::State& state) { EchoTestService::AsyncService service; @@ -642,6 +830,25 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair) BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) ->Range(0, 128 * 1024 * 1024); +static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) { + int msg_size = 0; + int num_ping_pongs = 0; + for (msg_size = 0; msg_size <= 128 * 1024 * 1024; + msg_size == 0 ? msg_size++ : msg_size *= 8) { + for (num_ping_pongs = 0; num_ping_pongs <= 2; num_ping_pongs++) { + b->Args({msg_size, num_ping_pongs}); + } + } +} + +BENCHMARK_TEMPLATE(BM_StreamingPingPong, InProcessCHTTP2, NoOpMutator, + NoOpMutator) + ->Apply(StreamingPingPongArgs); + +BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcessCHTTP2, NoOpMutator, + NoOpMutator) + ->Range(0, 128 * 1024 * 1024); + } // namespace testing } // namespace grpc From afaa75196e5f4098c60181020d307d72f1a580be Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Fri, 3 Feb 2017 19:43:47 -0800 Subject: [PATCH 2/4] Add comment --- test/cpp/microbenchmarks/bm_fullstack.cc | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index 611ee8ddb3a..e9259e154a3 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -830,6 +830,8 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair) BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) ->Range(0, 128 * 1024 * 1024); +// 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; int num_ping_pongs = 0; From 0b26e27056053f7e9651f849befdb466b96bea6e Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Fri, 3 Feb 2017 20:41:00 -0800 Subject: [PATCH 3/4] cr comments --- test/cpp/microbenchmarks/bm_fullstack.cc | 13 +++---------- 1 file changed, 3 insertions(+), 10 deletions(-) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index e9259e154a3..4981f5e27e9 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -491,8 +491,7 @@ static void BM_StreamingPingPong(benchmark::State& state) { } // Send 'max_ping_pongs' number of ping pong messages - int ping_pong_cnt = 0; - while (ping_pong_cnt < max_ping_pongs) { + while (state.iterations() < 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 @@ -511,8 +510,6 @@ static void BM_StreamingPingPong(benchmark::State& state) { GPR_ASSERT(need_tags & (1 << i)); need_tags &= ~(1 << i); } - - ping_pong_cnt++; } request_rw->WritesDone(tag(0)); @@ -543,7 +540,6 @@ static void BM_StreamingPingPong(benchmark::State& state) { template static void BM_StreamingPingPongMsgs(benchmark::State& state) { const int msg_size = state.range(0); - int ping_pong_cnt = 0; EchoTestService::AsyncService service; std::unique_ptr fixture(new Fixture(&service)); @@ -602,8 +598,6 @@ static void BM_StreamingPingPongMsgs(benchmark::State& state) { GPR_ASSERT(need_tags & (1 << i)); need_tags &= ~(1 << i); } - - ping_pong_cnt++; } request_rw->WritesDone(tag(0)); @@ -624,7 +618,7 @@ static void BM_StreamingPingPongMsgs(benchmark::State& state) { fixture->Finish(state); fixture.reset(); - state.SetBytesProcessed(msg_size * state.iterations() * ping_pong_cnt * 2); + state.SetBytesProcessed(msg_size * state.iterations() * 2); } template @@ -835,8 +829,7 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) { int msg_size = 0; int num_ping_pongs = 0; - for (msg_size = 0; msg_size <= 128 * 1024 * 1024; - msg_size == 0 ? msg_size++ : msg_size *= 8) { + for (msg_size = 1; msg_size <= 128 * 1024 * 1024; msg_size *= 8) { for (num_ping_pongs = 0; num_ping_pongs <= 2; num_ping_pongs++) { b->Args({msg_size, num_ping_pongs}); } From dfb82e076f3ed49312d2c1d6e28f4e667f8415c2 Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Fri, 3 Feb 2017 21:05:01 -0800 Subject: [PATCH 4/4] Address code review comments --- test/cpp/microbenchmarks/bm_fullstack.cc | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index 4981f5e27e9..dd180de48a5 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -491,7 +491,8 @@ static void BM_StreamingPingPong(benchmark::State& state) { } // Send 'max_ping_pongs' number of ping pong messages - while (state.iterations() < max_ping_pongs) { + 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 @@ -510,6 +511,8 @@ static void BM_StreamingPingPong(benchmark::State& state) { GPR_ASSERT(need_tags & (1 << i)); need_tags &= ~(1 << i); } + + ping_pong_cnt++; } request_rw->WritesDone(tag(0)); @@ -828,21 +831,27 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) // only "small streams" (i.e streams with 0, 1 or 2 messages) static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) { int msg_size = 0; - int num_ping_pongs = 0; - for (msg_size = 1; msg_size <= 128 * 1024 * 1024; msg_size *= 8) { - for (num_ping_pongs = 0; num_ping_pongs <= 2; num_ping_pongs++) { - b->Args({msg_size, num_ping_pongs}); - } + + 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_StreamingPingPongMsgs, TCP, NoOpMutator, NoOpMutator) + ->Range(0, 128 * 1024 * 1024); } // namespace testing } // namespace grpc