Fix QPS Async Client Next loop

pull/13177/head
Ken Payson 8 years ago
parent bf3f4be8f3
commit 918ce7a686
  1. 28
      test/cpp/qps/client_async.cc
  2. 3
      test/cpp/qps/server_async.cc

@ -245,9 +245,20 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
if (!cli_cqs_[cq_[thread_idx]]->Next(&got_tag, &ok)) { if (!cli_cqs_[cq_[thread_idx]]->Next(&got_tag, &ok)) {
return; return;
} }
ClientRpcContext* ctx; ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
std::mutex* shutdown_mu = &shutdown_state_[thread_idx]->mutex; std::mutex* shutdown_mu = &shutdown_state_[thread_idx]->mutex;
do { shutdown_mu->lock();
while (cli_cqs_[cq_[thread_idx]]->DoThenAsyncNext(
[&, ctx, ok, entry_ptr, shutdown_mu]() {
if (!ctx->RunNextState(ok, entry_ptr)) {
// The RPC and callback are done, so clone the ctx
// and kickstart the new one
ctx->StartNewClone(cli_cqs_[cq_[thread_idx]].get());
delete ctx;
}
shutdown_mu->unlock();
},
&got_tag, &ok, gpr_inf_future(GPR_CLOCK_REALTIME))) {
t->UpdateHistogram(entry_ptr); t->UpdateHistogram(entry_ptr);
// Got a regular event, so process it // Got a regular event, so process it
ctx = ClientRpcContext::detag(got_tag); ctx = ClientRpcContext::detag(got_tag);
@ -265,18 +276,7 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
shutdown_mu->unlock(); shutdown_mu->unlock();
return; return;
} }
} while (cli_cqs_[cq_[thread_idx]]->DoThenAsyncNext( }
[&, ctx, ok, entry_ptr, shutdown_mu]() {
bool next_ok = ok;
if (!ctx->RunNextState(next_ok, entry_ptr)) {
// The RPC and callback are done, so clone the ctx
// and kickstart the new one
ctx->StartNewClone(cli_cqs_[cq_[thread_idx]].get());
delete ctx;
}
shutdown_mu->unlock();
},
&got_tag, &ok, gpr_inf_future(GPR_CLOCK_REALTIME)));
} }
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_; std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;

@ -206,13 +206,12 @@ class AsyncQpsServerTest final : public grpc::testing::Server {
return; return;
} }
ServerRpcContext *ctx; ServerRpcContext *ctx;
std::mutex *mu_ptr; std::mutex *mu_ptr = &shutdown_state_[thread_idx]->mutex;
do { do {
ctx = detag(got_tag); ctx = detag(got_tag);
// The tag is a pointer to an RPC context to invoke // The tag is a pointer to an RPC context to invoke
// Proceed while holding a lock to make sure that // Proceed while holding a lock to make sure that
// this thread isn't supposed to shut down // this thread isn't supposed to shut down
mu_ptr = &shutdown_state_[thread_idx]->mutex;
mu_ptr->lock(); mu_ptr->lock();
if (shutdown_state_[thread_idx]->shutdown) { if (shutdown_state_[thread_idx]->shutdown) {
mu_ptr->unlock(); mu_ptr->unlock();

Loading…
Cancel
Save