test/cpp/end2end/thread_stress_test.cc - platform/external/grpc-grpc - Git at Google

 /*
  *
  * Copyright 2015 gRPC authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  *
  */

 #include <cinttypes>
 #include <mutex>
 #include <thread>

 #include <grpc/grpc.h>
 #include <grpc/support/time.h>
 #include <grpcpp/channel.h>
 #include <grpcpp/client_context.h>
 #include <grpcpp/create_channel.h>
 #include <grpcpp/resource_quota.h>
 #include <grpcpp/server.h>
 #include <grpcpp/server_builder.h>
 #include <grpcpp/server_context.h>

 #include "src/core/lib/surface/api_trace.h"
 #include "src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.h"
 #include "src/proto/grpc/testing/echo.grpc.pb.h"
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"

 #include <gtest/gtest.h>

 using grpc::testing::EchoRequest;
 using grpc::testing::EchoResponse;
 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 {
 namespace testing {

 class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
  public:
   TestServiceImpl() {}

   Status Echo(ServerContext* context, const EchoRequest* request,
               EchoResponse* response) override {
     response->set_message(request->message());
     return Status::OK;
   }
 };

 template <class Service>
 class CommonStressTest {
  public:
   CommonStressTest() : kMaxMessageSize_(8192) {}
   virtual ~CommonStressTest() {}
   virtual void SetUp() = 0;
   virtual void TearDown() = 0;
   virtual void ResetStub() = 0;
   virtual bool AllowExhaustion() = 0;
   grpc::testing::EchoTestService::Stub* GetStub() { return stub_.get(); }

  protected:
   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
   std::unique_ptr<Server> server_;

   virtual void SetUpStart(ServerBuilder* builder, Service* service) = 0;
   void SetUpStartCommon(ServerBuilder* builder, Service* service) {
     builder->RegisterService(service);
     builder->SetMaxMessageSize(
         kMaxMessageSize_);  // For testing max message size.
   }
   void SetUpEnd(ServerBuilder* builder) { server_ = builder->BuildAndStart(); }
   void TearDownStart() { server_->Shutdown(); }
   void TearDownEnd() {}

  private:
   const int kMaxMessageSize_;
 };

 template <class Service>
 class CommonStressTestInsecure : public CommonStressTest<Service> {
  public:
   void ResetStub() override {
     std::shared_ptr<Channel> channel =
         CreateChannel(server_address_.str(), InsecureChannelCredentials());
     this->stub_ = grpc::testing::EchoTestService::NewStub(channel);
   }
   bool AllowExhaustion() override { return false; }

  protected:
   void SetUpStart(ServerBuilder* builder, Service* service) override {
     int port = grpc_pick_unused_port_or_die();
     this->server_address_ << "localhost:" << port;
     // Setup server
     builder->AddListeningPort(server_address_.str(),
                               InsecureServerCredentials());
     this->SetUpStartCommon(builder, service);
   }

  private:
   std::ostringstream server_address_;
 };

 template <class Service, bool allow_resource_exhaustion>
 class CommonStressTestInproc : public CommonStressTest<Service> {
  public:
   void ResetStub() override {
     ChannelArguments args;
     std::shared_ptr<Channel> channel = this->server_->InProcessChannel(args);
     this->stub_ = grpc::testing::EchoTestService::NewStub(channel);
   }
   bool AllowExhaustion() override { return allow_resource_exhaustion; }

  protected:
   void SetUpStart(ServerBuilder* builder, Service* service) override {
     this->SetUpStartCommon(builder, service);
   }
 };

 template <class BaseClass>
 class CommonStressTestSyncServer : public BaseClass {
  public:
   void SetUp() override {
     ServerBuilder builder;
     this->SetUpStart(&builder, &service_);
     this->SetUpEnd(&builder);
   }
   void TearDown() override {
     this->TearDownStart();
     this->TearDownEnd();
   }

  private:
   TestServiceImpl service_;
 };

 template <class BaseClass>
 class CommonStressTestSyncServerLowThreadCount : public BaseClass {
  public:
   void SetUp() override {
     ServerBuilder builder;
     ResourceQuota quota;
     this->SetUpStart(&builder, &service_);
     quota.SetMaxThreads(4);
     builder.SetResourceQuota(quota);
     this->SetUpEnd(&builder);
   }
   void TearDown() override {
     this->TearDownStart();
     this->TearDownEnd();
   }

  private:
   TestServiceImpl service_;
 };

 template <class BaseClass>
 class CommonStressTestAsyncServer : public BaseClass {
  public:
   CommonStressTestAsyncServer() : contexts_(kNumAsyncServerThreads * 100) {}
   void SetUp() override {
     shutting_down_ = false;
     ServerBuilder builder;
     this->SetUpStart(&builder, &service_);
     cq_ = builder.AddCompletionQueue();
     this->SetUpEnd(&builder);
     for (int i = 0; i < kNumAsyncServerThreads * 100; i++) {
       RefreshContext(i);
     }
     for (int i = 0; i < kNumAsyncServerThreads; i++) {
       server_threads_.emplace_back(&CommonStressTestAsyncServer::ProcessRpcs,
                                    this);
     }
   }
   void TearDown() override {
     {
       std::unique_lock<std::mutex> l(mu_);
       this->TearDownStart();
       shutting_down_ = true;
       cq_->Shutdown();
     }

     for (int i = 0; i < kNumAsyncServerThreads; i++) {
       server_threads_[i].join();
     }

     void* ignored_tag;
     bool ignored_ok;
     while (cq_->Next(&ignored_tag, &ignored_ok))
       ;
     this->TearDownEnd();
   }

  private:
   void ProcessRpcs() {
     void* tag;
     bool ok;
     while (cq_->Next(&tag, &ok)) {
       if (ok) {
         int i = static_cast<int>(reinterpret_cast<intptr_t>(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) {
     std::unique_lock<std::mutex> 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<EchoResponse>(
               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*)static_cast<intptr_t>(i));
     }
   }
   struct Context {
     std::unique_ptr<ServerContext> srv_ctx;
     std::unique_ptr<grpc::ServerAsyncResponseWriter<EchoResponse>>
         response_writer;
     EchoRequest recv_request;
     enum { READY, DONE } state;
   };
   std::vector<Context> contexts_;
   ::grpc::testing::EchoTestService::AsyncService service_;
   std::unique_ptr<ServerCompletionQueue> cq_;
   bool shutting_down_;
   std::mutex mu_;
   std::vector<std::thread> server_threads_;
 };

 template <class Common>
 class End2endTest : public ::testing::Test {
  protected:
   End2endTest() {}
   void SetUp() override { common_.SetUp(); }
   void TearDown() override { common_.TearDown(); }
   void ResetStub() { common_.ResetStub(); }

   Common common_;
 };

 static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs,
                     bool allow_exhaustion, gpr_atm* errors) {
   EchoRequest request;
   EchoResponse response;
   request.set_message("Hello");

   for (int i = 0; i < num_rpcs; ++i) {
     ClientContext context;
     Status s = stub->Echo(&context, request, &response);
     EXPECT_TRUE(s.ok() || (allow_exhaustion &&
                            s.error_code() == StatusCode::RESOURCE_EXHAUSTED));
     if (!s.ok()) {
       if (!(allow_exhaustion &&
             s.error_code() == StatusCode::RESOURCE_EXHAUSTED)) {
         gpr_log(GPR_ERROR, "RPC error: %d: %s", s.error_code(),
                 s.error_message().c_str());
       }
       gpr_atm_no_barrier_fetch_add(errors, static_cast<gpr_atm>(1));
     } else {
       EXPECT_EQ(response.message(), request.message());
     }
   }
 }

 typedef ::testing::Types<
     CommonStressTestSyncServer<CommonStressTestInsecure<TestServiceImpl>>,
     CommonStressTestSyncServer<CommonStressTestInproc<TestServiceImpl, false>>,
     CommonStressTestSyncServerLowThreadCount<
         CommonStressTestInproc<TestServiceImpl, true>>,
     CommonStressTestAsyncServer<
         CommonStressTestInsecure<grpc::testing::EchoTestService::AsyncService>>,
     CommonStressTestAsyncServer<CommonStressTestInproc<
         grpc::testing::EchoTestService::AsyncService, false>>>
     CommonTypes;
 TYPED_TEST_CASE(End2endTest, CommonTypes);
 TYPED_TEST(End2endTest, ThreadStress) {
   this->common_.ResetStub();
   std::vector<std::thread> threads;
   gpr_atm errors;
   gpr_atm_rel_store(&errors, static_cast<gpr_atm>(0));
   threads.reserve(kNumThreads);
   for (int i = 0; i < kNumThreads; ++i) {
     threads.emplace_back(SendRpc, this->common_.GetStub(), kNumRpcs,
                          this->common_.AllowExhaustion(), &errors);
   }
   for (int i = 0; i < kNumThreads; ++i) {
     threads[i].join();
   }
   uint64_t error_cnt = static_cast<uint64_t>(gpr_atm_no_barrier_load(&errors));
   if (error_cnt != 0) {
     gpr_log(GPR_INFO, "RPC error count: %" PRIu64, error_cnt);
   }
   // If this test allows resource exhaustion, expect that it actually sees some
   if (this->common_.AllowExhaustion()) {
     EXPECT_GT(error_cnt, static_cast<uint64_t>(0));
   }
 }

 template <class Common>
 class AsyncClientEnd2endTest : public ::testing::Test {
  protected:
   AsyncClientEnd2endTest() : rpcs_outstanding_(0) {}

   void SetUp() override { common_.SetUp(); }
   void TearDown() override {
     void* ignored_tag;
     bool ignored_ok;
     while (cq_.Next(&ignored_tag, &ignored_ok))
       ;
     common_.TearDown();
   }

   void Wait() {
     std::unique_lock<std::mutex> l(mu_);
     while (rpcs_outstanding_ != 0) {
       cv_.wait(l);
     }

     cq_.Shutdown();
   }

   struct AsyncClientCall {
     EchoResponse response;
     ClientContext context;
     Status status;
     std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader;
   };

   void AsyncSendRpc(int num_rpcs) {
     for (int i = 0; i < num_rpcs; ++i) {
       AsyncClientCall* call = new AsyncClientCall;
       EchoRequest request;
       request.set_message("Hello: " + grpc::to_string(i));
       call->response_reader =
           common_.GetStub()->AsyncEcho(&call->context, request, &cq_);
       call->response_reader->Finish(&call->response, &call->status,
                                     (void*)call);

       std::unique_lock<std::mutex> l(mu_);
       rpcs_outstanding_++;
     }
   }

   void AsyncCompleteRpc() {
     while (true) {
       void* got_tag;
       bool ok = false;
       if (!cq_.Next(&got_tag, &ok)) break;
       AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);
       if (!ok) {
         gpr_log(GPR_DEBUG, "Error: %d", call->status.error_code());
       }
       delete call;

       bool notify;
       {
         std::unique_lock<std::mutex> l(mu_);
         rpcs_outstanding_--;
         notify = (rpcs_outstanding_ == 0);
       }
       if (notify) {
         cv_.notify_all();
       }
     }
   }

   Common common_;
   CompletionQueue cq_;
   std::mutex mu_;
   std::condition_variable cv_;
   int rpcs_outstanding_;
 };

 TYPED_TEST_CASE(AsyncClientEnd2endTest, CommonTypes);
 TYPED_TEST(AsyncClientEnd2endTest, ThreadStress) {
   this->common_.ResetStub();
   std::vector<std::thread> send_threads, completion_threads;
   for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
     completion_threads.emplace_back(
         &AsyncClientEnd2endTest_ThreadStress_Test<TypeParam>::AsyncCompleteRpc,
         this);
   }
   for (int i = 0; i < kNumAsyncSendThreads; ++i) {
     send_threads.emplace_back(
         &AsyncClientEnd2endTest_ThreadStress_Test<TypeParam>::AsyncSendRpc,
         this, kNumRpcs);
   }
   for (int i = 0; i < kNumAsyncSendThreads; ++i) {
     send_threads[i].join();
   }

   this->Wait();
   for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
     completion_threads[i].join();
   }
 }

 }  // namespace testing
 }  // namespace grpc

 int main(int argc, char** argv) {
   grpc::testing::TestEnvironment env(argc, argv);
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();
 }
	/*
	*
	* Copyright 2015 gRPC authors.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*
	*/

	#include <cinttypes>
	#include <mutex>
	#include <thread>

	#include <grpc/grpc.h>
	#include <grpc/support/time.h>
	#include <grpcpp/channel.h>
	#include <grpcpp/client_context.h>
	#include <grpcpp/create_channel.h>
	#include <grpcpp/resource_quota.h>
	#include <grpcpp/server.h>
	#include <grpcpp/server_builder.h>
	#include <grpcpp/server_context.h>

	#include "src/core/lib/surface/api_trace.h"
	#include "src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.h"
	#include "src/proto/grpc/testing/echo.grpc.pb.h"
	#include "test/core/util/port.h"
	#include "test/core/util/test_config.h"

	#include <gtest/gtest.h>

	using grpc::testing::EchoRequest;
	using grpc::testing::EchoResponse;
	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 {
	namespace testing {

	class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
	public:
	TestServiceImpl() {}

	Status Echo(ServerContext* context, const EchoRequest* request,
	EchoResponse* response) override {
	response->set_message(request->message());
	return Status::OK;
	}
	};

	template <class Service>
	class CommonStressTest {
	public:
	CommonStressTest() : kMaxMessageSize_(8192) {}
	virtual ~CommonStressTest() {}
	virtual void SetUp() = 0;
	virtual void TearDown() = 0;
	virtual void ResetStub() = 0;
	virtual bool AllowExhaustion() = 0;
	grpc::testing::EchoTestService::Stub* GetStub() { return stub_.get(); }

	protected:
	std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
	std::unique_ptr<Server> server_;

	virtual void SetUpStart(ServerBuilder* builder, Service* service) = 0;
	void SetUpStartCommon(ServerBuilder* builder, Service* service) {
	builder->RegisterService(service);
	builder->SetMaxMessageSize(
	kMaxMessageSize_); // For testing max message size.
	}
	void SetUpEnd(ServerBuilder* builder) { server_ = builder->BuildAndStart(); }
	void TearDownStart() { server_->Shutdown(); }
	void TearDownEnd() {}

	private:
	const int kMaxMessageSize_;
	};

	template <class Service>
	class CommonStressTestInsecure : public CommonStressTest<Service> {
	public:
	void ResetStub() override {
	std::shared_ptr<Channel> channel =
	CreateChannel(server_address_.str(), InsecureChannelCredentials());
	this->stub_ = grpc::testing::EchoTestService::NewStub(channel);
	}
	bool AllowExhaustion() override { return false; }

	protected:
	void SetUpStart(ServerBuilder* builder, Service* service) override {
	int port = grpc_pick_unused_port_or_die();
	this->server_address_ << "localhost:" << port;
	// Setup server
	builder->AddListeningPort(server_address_.str(),
	InsecureServerCredentials());
	this->SetUpStartCommon(builder, service);
	}

	private:
	std::ostringstream server_address_;
	};

	template <class Service, bool allow_resource_exhaustion>
	class CommonStressTestInproc : public CommonStressTest<Service> {
	public:
	void ResetStub() override {
	ChannelArguments args;
	std::shared_ptr<Channel> channel = this->server_->InProcessChannel(args);
	this->stub_ = grpc::testing::EchoTestService::NewStub(channel);
	}
	bool AllowExhaustion() override { return allow_resource_exhaustion; }

	protected:
	void SetUpStart(ServerBuilder* builder, Service* service) override {
	this->SetUpStartCommon(builder, service);
	}
	};

	template <class BaseClass>
	class CommonStressTestSyncServer : public BaseClass {
	public:
	void SetUp() override {
	ServerBuilder builder;
	this->SetUpStart(&builder, &service_);
	this->SetUpEnd(&builder);
	}
	void TearDown() override {
	this->TearDownStart();
	this->TearDownEnd();
	}

	private:
	TestServiceImpl service_;
	};

	template <class BaseClass>
	class CommonStressTestSyncServerLowThreadCount : public BaseClass {
	public:
	void SetUp() override {
	ServerBuilder builder;
	ResourceQuota quota;
	this->SetUpStart(&builder, &service_);
	quota.SetMaxThreads(4);
	builder.SetResourceQuota(quota);
	this->SetUpEnd(&builder);
	}
	void TearDown() override {
	this->TearDownStart();
	this->TearDownEnd();
	}

	private:
	TestServiceImpl service_;
	};

	template <class BaseClass>
	class CommonStressTestAsyncServer : public BaseClass {
	public:
	CommonStressTestAsyncServer() : contexts_(kNumAsyncServerThreads * 100) {}
	void SetUp() override {
	shutting_down_ = false;
	ServerBuilder builder;
	this->SetUpStart(&builder, &service_);
	cq_ = builder.AddCompletionQueue();
	this->SetUpEnd(&builder);
	for (int i = 0; i < kNumAsyncServerThreads * 100; i++) {
	RefreshContext(i);
	}
	for (int i = 0; i < kNumAsyncServerThreads; i++) {
	server_threads_.emplace_back(&CommonStressTestAsyncServer::ProcessRpcs,
	this);
	}
	}
	void TearDown() override {
	{
	std::unique_lock<std::mutex> l(mu_);
	this->TearDownStart();
	shutting_down_ = true;
	cq_->Shutdown();
	}

	for (int i = 0; i < kNumAsyncServerThreads; i++) {
	server_threads_[i].join();
	}

	void* ignored_tag;
	bool ignored_ok;
	while (cq_->Next(&ignored_tag, &ignored_ok))
	;
	this->TearDownEnd();
	}

	private:
	void ProcessRpcs() {
	void* tag;
	bool ok;
	while (cq_->Next(&tag, &ok)) {
	if (ok) {
	int i = static_cast<int>(reinterpret_cast<intptr_t>(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) {
	std::unique_lock<std::mutex> 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<EchoResponse>(
	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*)static_cast<intptr_t>(i));
	}
	}
	struct Context {
	std::unique_ptr<ServerContext> srv_ctx;
	std::unique_ptr<grpc::ServerAsyncResponseWriter<EchoResponse>>
	response_writer;
	EchoRequest recv_request;
	enum { READY, DONE } state;
	};
	std::vector<Context> contexts_;
	::grpc::testing::EchoTestService::AsyncService service_;
	std::unique_ptr<ServerCompletionQueue> cq_;
	bool shutting_down_;
	std::mutex mu_;
	std::vector<std::thread> server_threads_;
	};

	template <class Common>
	class End2endTest : public ::testing::Test {
	protected:
	End2endTest() {}
	void SetUp() override { common_.SetUp(); }
	void TearDown() override { common_.TearDown(); }
	void ResetStub() { common_.ResetStub(); }

	Common common_;
	};

	static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs,
	bool allow_exhaustion, gpr_atm* errors) {
	EchoRequest request;
	EchoResponse response;
	request.set_message("Hello");

	for (int i = 0; i < num_rpcs; ++i) {
	ClientContext context;
	Status s = stub->Echo(&context, request, &response);
	EXPECT_TRUE(s.ok() \|\| (allow_exhaustion &&
	s.error_code() == StatusCode::RESOURCE_EXHAUSTED));
	if (!s.ok()) {
	if (!(allow_exhaustion &&
	s.error_code() == StatusCode::RESOURCE_EXHAUSTED)) {
	gpr_log(GPR_ERROR, "RPC error: %d: %s", s.error_code(),
	s.error_message().c_str());
	}
	gpr_atm_no_barrier_fetch_add(errors, static_cast<gpr_atm>(1));
	} else {
	EXPECT_EQ(response.message(), request.message());
	}
	}
	}

	typedef ::testing::Types<
	CommonStressTestSyncServer<CommonStressTestInsecure<TestServiceImpl>>,
	CommonStressTestSyncServer<CommonStressTestInproc<TestServiceImpl, false>>,
	CommonStressTestSyncServerLowThreadCount<
	CommonStressTestInproc<TestServiceImpl, true>>,
	CommonStressTestAsyncServer<
	CommonStressTestInsecure<grpc::testing::EchoTestService::AsyncService>>,
	CommonStressTestAsyncServer<CommonStressTestInproc<
	grpc::testing::EchoTestService::AsyncService, false>>>
	CommonTypes;
	TYPED_TEST_CASE(End2endTest, CommonTypes);
	TYPED_TEST(End2endTest, ThreadStress) {
	this->common_.ResetStub();
	std::vector<std::thread> threads;
	gpr_atm errors;
	gpr_atm_rel_store(&errors, static_cast<gpr_atm>(0));
	threads.reserve(kNumThreads);
	for (int i = 0; i < kNumThreads; ++i) {
	threads.emplace_back(SendRpc, this->common_.GetStub(), kNumRpcs,
	this->common_.AllowExhaustion(), &errors);
	}
	for (int i = 0; i < kNumThreads; ++i) {
	threads[i].join();
	}
	uint64_t error_cnt = static_cast<uint64_t>(gpr_atm_no_barrier_load(&errors));
	if (error_cnt != 0) {
	gpr_log(GPR_INFO, "RPC error count: %" PRIu64, error_cnt);
	}
	// If this test allows resource exhaustion, expect that it actually sees some
	if (this->common_.AllowExhaustion()) {
	EXPECT_GT(error_cnt, static_cast<uint64_t>(0));
	}
	}

	template <class Common>
	class AsyncClientEnd2endTest : public ::testing::Test {
	protected:
	AsyncClientEnd2endTest() : rpcs_outstanding_(0) {}

	void SetUp() override { common_.SetUp(); }
	void TearDown() override {
	void* ignored_tag;
	bool ignored_ok;
	while (cq_.Next(&ignored_tag, &ignored_ok))
	;
	common_.TearDown();
	}

	void Wait() {
	std::unique_lock<std::mutex> l(mu_);
	while (rpcs_outstanding_ != 0) {
	cv_.wait(l);
	}

	cq_.Shutdown();
	}

	struct AsyncClientCall {
	EchoResponse response;
	ClientContext context;
	Status status;
	std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader;
	};

	void AsyncSendRpc(int num_rpcs) {
	for (int i = 0; i < num_rpcs; ++i) {
	AsyncClientCall* call = new AsyncClientCall;
	EchoRequest request;
	request.set_message("Hello: " + grpc::to_string(i));
	call->response_reader =
	common_.GetStub()->AsyncEcho(&call->context, request, &cq_);
	call->response_reader->Finish(&call->response, &call->status,
	(void*)call);

	std::unique_lock<std::mutex> l(mu_);
	rpcs_outstanding_++;
	}
	}

	void AsyncCompleteRpc() {
	while (true) {
	void* got_tag;
	bool ok = false;
	if (!cq_.Next(&got_tag, &ok)) break;
	AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);
	if (!ok) {
	gpr_log(GPR_DEBUG, "Error: %d", call->status.error_code());
	}
	delete call;

	bool notify;
	{
	std::unique_lock<std::mutex> l(mu_);
	rpcs_outstanding_--;
	notify = (rpcs_outstanding_ == 0);
	}
	if (notify) {
	cv_.notify_all();
	}
	}
	}

	Common common_;
	CompletionQueue cq_;
	std::mutex mu_;
	std::condition_variable cv_;
	int rpcs_outstanding_;
	};

	TYPED_TEST_CASE(AsyncClientEnd2endTest, CommonTypes);
	TYPED_TEST(AsyncClientEnd2endTest, ThreadStress) {
	this->common_.ResetStub();
	std::vector<std::thread> send_threads, completion_threads;
	for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
	completion_threads.emplace_back(
	&AsyncClientEnd2endTest_ThreadStress_Test<TypeParam>::AsyncCompleteRpc,
	this);
	}
	for (int i = 0; i < kNumAsyncSendThreads; ++i) {
	send_threads.emplace_back(
	&AsyncClientEnd2endTest_ThreadStress_Test<TypeParam>::AsyncSendRpc,
	this, kNumRpcs);
	}
	for (int i = 0; i < kNumAsyncSendThreads; ++i) {
	send_threads[i].join();
	}

	this->Wait();
	for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
	completion_threads[i].join();
	}
	}

	} // namespace testing
	} // namespace grpc

	int main(int argc, char** argv) {
	grpc::testing::TestEnvironment env(argc, argv);
	::testing::InitGoogleTest(&argc, argv);
	return RUN_ALL_TESTS();
	}