From e44d1a634af323e6b0270a02cf8c0d61ca8fc41e Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 27 Jan 2017 17:06:16 -0800 Subject: [PATCH 1/4] Add client->server long-lived streaming --- test/cpp/microbenchmarks/bm_fullstack.cc | 62 ++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index fa0d2f1fb9e..76d8fcffcd8 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -439,6 +439,63 @@ static void BM_UnaryPingPong(benchmark::State& state) { state.range(1) * state.iterations()); } +template +static void BM_PumpStreamClientToServer(benchmark::State& state) { + EchoTestService::AsyncService service; + std::unique_ptr fixture(new Fixture(&service)); + { + EchoRequest send_request; + EchoRequest recv_request; + if (state.range(0) > 0) { + send_request.set_message(std::string(state.range(0), 'a')); + } + Status recv_status; + ServerContext svr_ctx; + ServerAsyncReaderWriter response_rw(&svr_ctx); + service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(), + fixture->cq(), tag(0)); + std::unique_ptr 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()) { + request_rw->Write(send_request, tag(1)); + while (true) { + 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); + } + } + } + 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); + } + } + fixture->Finish(state); + fixture.reset(); + state.SetBytesProcessed(state.range(0) * state.iterations()); +} + /******************************************************************************* * CONFIGURATIONS */ @@ -511,6 +568,11 @@ BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator, Server_AddInitialMetadata, 100>) ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, TCP) + ->Range(0, 128 * 1024 * 1024); +BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, InProcessCHTTP2) + ->Range(0, 128 * 1024 * 1024); + } // namespace testing } // namespace grpc From db13ac408434ccdab30937293fe788e1ef3fc3b7 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 27 Jan 2017 17:11:38 -0800 Subject: [PATCH 2/4] Add server->client long-lived streaming --- test/cpp/microbenchmarks/bm_fullstack.cc | 61 ++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index 76d8fcffcd8..7ba73080799 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -496,6 +496,63 @@ static void BM_PumpStreamClientToServer(benchmark::State& state) { state.SetBytesProcessed(state.range(0) * state.iterations()); } +template +static void BM_PumpStreamServerToClient(benchmark::State& state) { + EchoTestService::AsyncService service; + std::unique_ptr 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')); + } + Status recv_status; + ServerContext svr_ctx; + ServerAsyncReaderWriter response_rw(&svr_ctx); + service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(), + fixture->cq(), tag(0)); + std::unique_ptr 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); + } + request_rw->Read(&recv_response, tag(0)); + while (state.KeepRunning()) { + response_rw.Write(send_response, tag(1)); + while (true) { + 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); + } + } + } + response_rw.Finish(Status::OK, 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); + } + } + fixture->Finish(state); + fixture.reset(); + state.SetBytesProcessed(state.range(0) * state.iterations()); +} + /******************************************************************************* * CONFIGURATIONS */ @@ -572,6 +629,10 @@ BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, TCP) ->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, InProcessCHTTP2) + ->Range(0, 128 * 1024 * 1024); } // namespace testing } // namespace grpc From cf14a1bd1e137facd58c427d851eccb0c64944d2 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 27 Jan 2017 17:18:47 -0800 Subject: [PATCH 3/4] Add other fixtures to pump tests --- test/cpp/microbenchmarks/bm_fullstack.cc | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc index 7ba73080799..c3e96c572c8 100644 --- a/test/cpp/microbenchmarks/bm_fullstack.cc +++ b/test/cpp/microbenchmarks/bm_fullstack.cc @@ -627,10 +627,18 @@ BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator, 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) + ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) ->Range(0, 128 * 1024 * 1024); From e985cf57c24a15f50906e44cb93c067711f78be7 Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Thu, 2 Feb 2017 19:19:01 -0800 Subject: [PATCH 4/4] Fixed invalid handling of slice data in interop_server --- test/cpp/interop/interop_server.cc | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/cpp/interop/interop_server.cc b/test/cpp/interop/interop_server.cc index 77e309dde49..5a810b45ef4 100644 --- a/test/cpp/interop/interop_server.cc +++ b/test/cpp/interop/interop_server.cc @@ -106,7 +106,9 @@ void MaybeEchoMetadata(ServerContext* context) { if (iter != client_metadata.end()) { iter = client_metadata.find("user-agent"); if (iter != client_metadata.end()) { - context->AddInitialMetadata(kEchoUserAgentKey, iter->second.data()); + context->AddInitialMetadata( + kEchoUserAgentKey, + grpc::string(iter->second.begin(), iter->second.end())); } } }