From bbbbf621e408bd8f744bd92dacedaa04771364cd Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 29 Mar 2016 00:35:26 -0700 Subject: [PATCH 1/3] Add a concurrent test for sync client, async server case --- test/cpp/end2end/thread_stress_test.cc | 162 ++++++++++++++++++++++--- 1 file changed, 144 insertions(+), 18 deletions(-) diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc index 8760b8d28e3..ff20292812f 100644 --- a/test/cpp/end2end/thread_stress_test.cc +++ b/test/cpp/end2end/thread_stress_test.cc @@ -58,6 +58,7 @@ using std::chrono::system_clock; const int kNumThreads = 100; // Number of threads const int kNumAsyncSendThreads = 2; const int kNumAsyncReceiveThreads = 50; +const int kNumAsyncServerThreads = 50; const int kNumRpcs = 1000; // Number of RPCs per thread namespace grpc { @@ -174,39 +175,142 @@ class TestServiceImplDupPkg } }; +template class CommonStressTest { public: CommonStressTest() : kMaxMessageSize_(8192) {} - void SetUp() { - int port = grpc_pick_unused_port_or_die(); - server_address_ << "localhost:" << port; - // Setup server - ServerBuilder builder; - builder.AddListeningPort(server_address_.str(), - InsecureServerCredentials()); - builder.RegisterService(&service_); - builder.SetMaxMessageSize( - kMaxMessageSize_); // For testing max message size. - builder.RegisterService(&dup_pkg_service_); - server_ = builder.BuildAndStart(); - } - void TearDown() { server_->Shutdown(); } + virtual void SetUp() = 0; + virtual void TearDown() = 0; void ResetStub() { std::shared_ptr channel = CreateChannel(server_address_.str(), InsecureChannelCredentials()); stub_ = grpc::testing::EchoTestService::NewStub(channel); } grpc::testing::EchoTestService::Stub* GetStub() { return stub_.get(); } - + protected: + void SetUpStart(ServerBuilder *builder, Service *service) { + int port = grpc_pick_unused_port_or_die(); + server_address_ << "localhost:" << port; + // Setup server + builder->AddListeningPort(server_address_.str(), + InsecureServerCredentials()); + builder->RegisterService(service); + builder->SetMaxMessageSize( + kMaxMessageSize_); // For testing max message size. + builder->RegisterService(&dup_pkg_service_); + } + void SetUpEnd(ServerBuilder *builder) { + server_ = builder->BuildAndStart(); + } + void TearDownStart() { server_->Shutdown(); } + void TearDownEnd() { } private: std::unique_ptr stub_; std::unique_ptr server_; std::ostringstream server_address_; const int kMaxMessageSize_; - TestServiceImpl service_; TestServiceImplDupPkg dup_pkg_service_; }; +class CommonStressTestSyncServer : public CommonStressTest { + public: + void SetUp() GRPC_OVERRIDE { + ServerBuilder builder; + SetUpStart(&builder, &service_); + SetUpEnd(&builder); + } + void TearDown() GRPC_OVERRIDE { + TearDownStart(); + TearDownEnd(); + } + private: + TestServiceImpl service_; +}; + +class CommonStressTestAsyncServer : + public CommonStressTest< ::grpc::testing::EchoTestService::AsyncService> { + public: + void SetUp() GRPC_OVERRIDE { + shutting_down_ = false; + ServerBuilder builder; + SetUpStart(&builder, &service_); + cq_ = builder.AddCompletionQueue(); + SetUpEnd(&builder); + contexts_ = new Context[kNumAsyncServerThreads * 100]; + for (int i = 0; i < kNumAsyncServerThreads * 100; i++) { + RefreshContext(i); + } + for (int i = 0; i < kNumAsyncServerThreads; i++) { + server_threads_.push_back(new std::thread(&CommonStressTestAsyncServer::ProcessRpcs, this)); + } + } + void TearDown() GRPC_OVERRIDE { + { + unique_lock l(mu_); + TearDownStart(); + shutting_down_ = true; + cq_->Shutdown(); + } + + for (int i = 0; i < kNumAsyncServerThreads; i++) { + server_threads_[i]->join(); + delete server_threads_[i]; + } + + void* ignored_tag; + bool ignored_ok; + while (cq_->Next(&ignored_tag, &ignored_ok)) + ; + TearDownEnd(); + delete[] contexts_; + } + private: + void ProcessRpcs() { + void *tag; + bool ok; + while (cq_->Next(&tag, &ok)) { + if (ok) { + int i = static_cast(reinterpret_cast(tag)); + switch (contexts_[i].state) { + case Context::READY: { + contexts_[i].state = Context::DONE; + EchoResponse send_response; + send_response.set_message(contexts_[i].recv_request.message()); + contexts_[i].response_writer->Finish(send_response, Status::OK, tag); + break; + } + case Context::DONE: + RefreshContext(i); + break; + } + } + } + } + void RefreshContext(int i) { + unique_lock l(mu_); + if (!shutting_down_) { + contexts_[i].state = Context::READY; + contexts_[i].srv_ctx.reset(new ServerContext); + contexts_[i].response_writer.reset(new grpc::ServerAsyncResponseWriter(contexts_[i].srv_ctx.get())); + service_.RequestEcho(contexts_[i].srv_ctx.get(), &contexts_[i].recv_request, + contexts_[i].response_writer.get(), cq_.get(), + cq_.get(), (void*)(intptr_t)i); + } + } + struct Context { + std::unique_ptr srv_ctx; + std::unique_ptr> + response_writer; + EchoRequest recv_request; + enum {READY, DONE} state; + } *contexts_; + ::grpc::testing::EchoTestService::AsyncService service_; + std::unique_ptr cq_; + bool shutting_down_; + mutex mu_; + std::vector server_threads_; +}; + class End2endTest : public ::testing::Test { protected: End2endTest() {} @@ -214,7 +318,17 @@ class End2endTest : public ::testing::Test { void TearDown() GRPC_OVERRIDE { common_.TearDown(); } void ResetStub() { common_.ResetStub(); } - CommonStressTest common_; + CommonStressTestSyncServer common_; +}; + +class End2endTestAsyncServer : public ::testing::Test { + protected: + End2endTestAsyncServer() {} + void SetUp() GRPC_OVERRIDE { common_.SetUp(); } + void TearDown() GRPC_OVERRIDE { common_.TearDown(); } + void ResetStub() { common_.ResetStub(); } + + CommonStressTestAsyncServer common_; }; static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs) { @@ -242,6 +356,18 @@ TEST_F(End2endTest, ThreadStress) { } } +TEST_F(End2endTestAsyncServer, ThreadStress) { + common_.ResetStub(); + std::vector threads; + for (int i = 0; i < kNumThreads; ++i) { + threads.push_back(new std::thread(SendRpc, common_.GetStub(), kNumRpcs)); + } + for (int i = 0; i < kNumThreads; ++i) { + threads[i]->join(); + delete threads[i]; + } +} + class AsyncClientEnd2endTest : public ::testing::Test { protected: AsyncClientEnd2endTest() : rpcs_outstanding_(0) {} @@ -309,7 +435,7 @@ class AsyncClientEnd2endTest : public ::testing::Test { } } - CommonStressTest common_; + CommonStressTestSyncServer common_; CompletionQueue cq_; mutex mu_; condition_variable cv_; From 40d1a2cb876e3ee19bccd5a59f3cfba48a22c4a5 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 29 Mar 2016 00:38:09 -0700 Subject: [PATCH 2/3] clang-format --- test/cpp/end2end/thread_stress_test.cc | 39 +++++++++++++++----------- 1 file changed, 23 insertions(+), 16 deletions(-) diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc index ff20292812f..22ea3a82b4a 100644 --- a/test/cpp/end2end/thread_stress_test.cc +++ b/test/cpp/end2end/thread_stress_test.cc @@ -187,8 +187,9 @@ class CommonStressTest { stub_ = grpc::testing::EchoTestService::NewStub(channel); } grpc::testing::EchoTestService::Stub* GetStub() { return stub_.get(); } + protected: - void SetUpStart(ServerBuilder *builder, Service *service) { + void SetUpStart(ServerBuilder* builder, Service* service) { int port = grpc_pick_unused_port_or_die(); server_address_ << "localhost:" << port; // Setup server @@ -199,11 +200,10 @@ class CommonStressTest { kMaxMessageSize_); // For testing max message size. builder->RegisterService(&dup_pkg_service_); } - void SetUpEnd(ServerBuilder *builder) { - server_ = builder->BuildAndStart(); - } + void SetUpEnd(ServerBuilder* builder) { server_ = builder->BuildAndStart(); } void TearDownStart() { server_->Shutdown(); } - void TearDownEnd() { } + void TearDownEnd() {} + private: std::unique_ptr stub_; std::unique_ptr server_; @@ -223,12 +223,13 @@ class CommonStressTestSyncServer : public CommonStressTest { TearDownStart(); TearDownEnd(); } + private: TestServiceImpl service_; }; -class CommonStressTestAsyncServer : - public CommonStressTest< ::grpc::testing::EchoTestService::AsyncService> { +class CommonStressTestAsyncServer + : public CommonStressTest<::grpc::testing::EchoTestService::AsyncService> { public: void SetUp() GRPC_OVERRIDE { shutting_down_ = false; @@ -241,7 +242,8 @@ class CommonStressTestAsyncServer : RefreshContext(i); } for (int i = 0; i < kNumAsyncServerThreads; i++) { - server_threads_.push_back(new std::thread(&CommonStressTestAsyncServer::ProcessRpcs, this)); + server_threads_.push_back( + new std::thread(&CommonStressTestAsyncServer::ProcessRpcs, this)); } } void TearDown() GRPC_OVERRIDE { @@ -264,9 +266,10 @@ class CommonStressTestAsyncServer : TearDownEnd(); delete[] contexts_; } + private: void ProcessRpcs() { - void *tag; + void* tag; bool ok; while (cq_->Next(&tag, &ok)) { if (ok) { @@ -276,7 +279,8 @@ class CommonStressTestAsyncServer : contexts_[i].state = Context::DONE; EchoResponse send_response; send_response.set_message(contexts_[i].recv_request.message()); - contexts_[i].response_writer->Finish(send_response, Status::OK, tag); + contexts_[i].response_writer->Finish(send_response, Status::OK, + tag); break; } case Context::DONE: @@ -291,8 +295,11 @@ class CommonStressTestAsyncServer : if (!shutting_down_) { contexts_[i].state = Context::READY; contexts_[i].srv_ctx.reset(new ServerContext); - contexts_[i].response_writer.reset(new grpc::ServerAsyncResponseWriter(contexts_[i].srv_ctx.get())); - service_.RequestEcho(contexts_[i].srv_ctx.get(), &contexts_[i].recv_request, + contexts_[i].response_writer.reset( + new grpc::ServerAsyncResponseWriter( + contexts_[i].srv_ctx.get())); + service_.RequestEcho(contexts_[i].srv_ctx.get(), + &contexts_[i].recv_request, contexts_[i].response_writer.get(), cq_.get(), cq_.get(), (void*)(intptr_t)i); } @@ -300,15 +307,15 @@ class CommonStressTestAsyncServer : struct Context { std::unique_ptr srv_ctx; std::unique_ptr> - response_writer; + response_writer; EchoRequest recv_request; - enum {READY, DONE} state; - } *contexts_; + enum { READY, DONE } state; + } * contexts_; ::grpc::testing::EchoTestService::AsyncService service_; std::unique_ptr cq_; bool shutting_down_; mutex mu_; - std::vector server_threads_; + std::vector server_threads_; }; class End2endTest : public ::testing::Test { From 98f2f754b8b4cf460720eaf605c55734f28b2057 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 29 Mar 2016 00:54:11 -0700 Subject: [PATCH 3/3] Used TYPED_TEST to parametrize Include all 4 sync/async client/server combos --- test/cpp/end2end/thread_stress_test.cc | 55 ++++++++++---------------- 1 file changed, 21 insertions(+), 34 deletions(-) diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc index 22ea3a82b4a..3f75a0c92eb 100644 --- a/test/cpp/end2end/thread_stress_test.cc +++ b/test/cpp/end2end/thread_stress_test.cc @@ -318,6 +318,7 @@ class CommonStressTestAsyncServer std::vector server_threads_; }; +template class End2endTest : public ::testing::Test { protected: End2endTest() {} @@ -325,17 +326,7 @@ class End2endTest : public ::testing::Test { void TearDown() GRPC_OVERRIDE { common_.TearDown(); } void ResetStub() { common_.ResetStub(); } - CommonStressTestSyncServer common_; -}; - -class End2endTestAsyncServer : public ::testing::Test { - protected: - End2endTestAsyncServer() {} - void SetUp() GRPC_OVERRIDE { common_.SetUp(); } - void TearDown() GRPC_OVERRIDE { common_.TearDown(); } - void ResetStub() { common_.ResetStub(); } - - CommonStressTestAsyncServer common_; + Common common_; }; static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs) { @@ -351,23 +342,16 @@ static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs) { } } -TEST_F(End2endTest, ThreadStress) { - common_.ResetStub(); - std::vector threads; - for (int i = 0; i < kNumThreads; ++i) { - threads.push_back(new std::thread(SendRpc, common_.GetStub(), kNumRpcs)); - } - for (int i = 0; i < kNumThreads; ++i) { - threads[i]->join(); - delete threads[i]; - } -} - -TEST_F(End2endTestAsyncServer, ThreadStress) { - common_.ResetStub(); +typedef ::testing::Types + CommonTypes; +TYPED_TEST_CASE(End2endTest, CommonTypes); +TYPED_TEST(End2endTest, ThreadStress) { + this->common_.ResetStub(); std::vector threads; for (int i = 0; i < kNumThreads; ++i) { - threads.push_back(new std::thread(SendRpc, common_.GetStub(), kNumRpcs)); + threads.push_back( + new std::thread(SendRpc, this->common_.GetStub(), kNumRpcs)); } for (int i = 0; i < kNumThreads; ++i) { threads[i]->join(); @@ -375,6 +359,7 @@ TEST_F(End2endTestAsyncServer, ThreadStress) { } } +template class AsyncClientEnd2endTest : public ::testing::Test { protected: AsyncClientEnd2endTest() : rpcs_outstanding_(0) {} @@ -442,31 +427,33 @@ class AsyncClientEnd2endTest : public ::testing::Test { } } - CommonStressTestSyncServer common_; + Common common_; CompletionQueue cq_; mutex mu_; condition_variable cv_; int rpcs_outstanding_; }; -TEST_F(AsyncClientEnd2endTest, ThreadStress) { - common_.ResetStub(); +TYPED_TEST_CASE(AsyncClientEnd2endTest, CommonTypes); +TYPED_TEST(AsyncClientEnd2endTest, ThreadStress) { + this->common_.ResetStub(); std::vector send_threads, completion_threads; for (int i = 0; i < kNumAsyncReceiveThreads; ++i) { completion_threads.push_back(new std::thread( - &AsyncClientEnd2endTest_ThreadStress_Test::AsyncCompleteRpc, this)); + &AsyncClientEnd2endTest_ThreadStress_Test::AsyncCompleteRpc, + this)); } for (int i = 0; i < kNumAsyncSendThreads; ++i) { - send_threads.push_back( - new std::thread(&AsyncClientEnd2endTest_ThreadStress_Test::AsyncSendRpc, - this, kNumRpcs)); + send_threads.push_back(new std::thread( + &AsyncClientEnd2endTest_ThreadStress_Test::AsyncSendRpc, + this, kNumRpcs)); } for (int i = 0; i < kNumAsyncSendThreads; ++i) { send_threads[i]->join(); delete send_threads[i]; } - Wait(); + this->Wait(); for (int i = 0; i < kNumAsyncReceiveThreads; ++i) { completion_threads[i]->join(); delete completion_threads[i];