1 files changed, 83 insertions, 223 deletions

diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 9ea9cfe8b9..5eb9ff6521 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -49,86 +49,53 @@
 #include "test/core/util/grpc_profiler.h"
 #include "test/cpp/util/create_test_channel.h"
 #include "test/cpp/qps/qpstest.pb.h"
+#include "test/cpp/qps/timer.h"
+#include "test/cpp/qps/client.h"
 
-DEFINE_bool(enable_ssl, false, "Whether to use ssl/tls.");
-DEFINE_int32(server_port, 0, "Server port.");
-DEFINE_string(server_host, "127.0.0.1", "Server host.");
-DEFINE_int32(client_threads, 4, "Number of client threads.");
-
-// We have a configurable number of channels for sending RPCs.
-// RPCs are sent round-robin on the available channels by the
-// various threads. Interesting cases are 1 global channel or
-// 1 per-thread channel, but we can support any number.
-// The channels are assigned round-robin on an RPC by RPC basis
-// rather than just at initialization time in order to also measure the
-// impact of cache thrashing caused by channel changes. This is an issue
-// if you are not in one of the above "interesting cases"
-DEFINE_int32(client_channels, 4, "Number of client channels.");
-
-DEFINE_int32(num_rpcs, 1000, "Number of RPCs per thread.");
-DEFINE_int32(payload_size, 1, "Payload size in bytes");
-
-// Alternatively, specify parameters for test as a workload so that multiple
-// tests are initiated back-to-back. This is convenient for keeping a borg
-// allocation consistent. This is a space-separated list of
-// [threads channels num_rpcs payload_size ]*
-DEFINE_string(workload, "", "Workload parameters");
-
-using grpc::ChannelInterface;
-using grpc::CreateTestChannel;
-using grpc::testing::ServerStats;
-using grpc::testing::SimpleRequest;
-using grpc::testing::SimpleResponse;
-using grpc::testing::StatsRequest;
-using grpc::testing::TestService;
-
-// In some distros, gflags is in the namespace google, and in some others,
-// in gflags. This hack is enabling us to find both.
-namespace google {}
-namespace gflags {}
-using namespace google;
-using namespace gflags;
-
-static double now() {
-  gpr_timespec tv = gpr_now();
-  return 1e9 * tv.tv_sec + tv.tv_nsec;
-}
+namespace grpc {
+namespace testing {
 
 class ClientRpcContext {
  public:
   ClientRpcContext() {}
   virtual ~ClientRpcContext() {}
   virtual bool RunNextState() = 0;  // do next state, return false if steps done
+  virtual void StartNewClone() = 0;
   static void *tag(ClientRpcContext *c) { return reinterpret_cast<void *>(c); }
   static ClientRpcContext *detag(void *t) {
     return reinterpret_cast<ClientRpcContext *>(t);
   }
-  virtual void report_stats(gpr_histogram *hist) = 0;
+  virtual void report_stats(Histogram *hist) = 0;
 };
+
 template <class RequestType, class ResponseType>
 class ClientRpcContextUnaryImpl : public ClientRpcContext {
  public:
   ClientRpcContextUnaryImpl(
-      TestService::Stub *stub,
-      const RequestType &req,
+      TestService::Stub *stub, const RequestType &req,
       std::function<
           std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
-	      TestService::Stub *, grpc::ClientContext *, const RequestType &,
-	      void *)> start_req,
+              TestService::Stub *, grpc::ClientContext *, const RequestType &,
+              void *)> start_req,
       std::function<void(grpc::Status, ResponseType *)> on_done)
       : context_(),
-	stub_(stub),
+        stub_(stub),
         req_(req),
         response_(),
         next_state_(&ClientRpcContextUnaryImpl::ReqSent),
         callback_(on_done),
-        start_(now()),
+        start_req_(start_req),
+        start_(Timer::Now()),
         response_reader_(
-	    start_req(stub_, &context_, req_, ClientRpcContext::tag(this))) {}
+            start_req(stub_, &context_, req_, ClientRpcContext::tag(this))) {}
   ~ClientRpcContextUnaryImpl() GRPC_OVERRIDE {}
   bool RunNextState() GRPC_OVERRIDE { return (this->*next_state_)(); }
-  void report_stats(gpr_histogram *hist) GRPC_OVERRIDE {
-    gpr_histogram_add(hist, now() - start_);
+  void report_stats(Histogram *hist) GRPC_OVERRIDE {
+    hist->Add((Timer::Now() - start_) * 1e9);
+  }
+
+  void StartNewClone() {
+    new ClientRpcContextUnaryImpl(stub_, req_, start_req_, callback_);
   }
 
  private:
@@ -151,191 +118,84 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
   ResponseType response_;
   bool (ClientRpcContextUnaryImpl::*next_state_)();
   std::function<void(grpc::Status, ResponseType *)> callback_;
+  std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
+      TestService::Stub *, grpc::ClientContext *, const RequestType &, void *)>
+      start_req_;
   grpc::Status status_;
   double start_;
   std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>
       response_reader_;
 };
 
-static void RunTest(const int client_threads, const int client_channels,
-                    const int num_rpcs, const int payload_size) {
-  gpr_log(GPR_INFO,
-          "QPS test with parameters\n"
-          "enable_ssl = %d\n"
-          "client_channels = %d\n"
-          "client_threads = %d\n"
-          "num_rpcs = %d\n"
-          "payload_size = %d\n"
-          "server_host:server_port = %s:%d\n\n",
-          FLAGS_enable_ssl, client_channels, client_threads, num_rpcs,
-          payload_size, FLAGS_server_host.c_str(), FLAGS_server_port);
-
-  std::ostringstream oss;
-  oss << FLAGS_server_host << ":" << FLAGS_server_port;
-
-  class ClientChannelInfo {
-   public:
-    explicit ClientChannelInfo(const grpc::string &server)
-        : channel_(CreateTestChannel(server, FLAGS_enable_ssl)),
-          stub_(TestService::NewStub(channel_)) {}
-    ChannelInterface *get_channel() { return channel_.get(); }
-    TestService::Stub *get_stub() { return stub_.get(); }
+class AsyncClient GRPC_FINAL : public Client {
+ public:
+  explicit AsyncClient(const ClientConfig &config) : Client(config) {
+    for (int i = 0; i < config.async_client_threads(); i++) {
+      cli_cqs_.emplace_back(new CompletionQueue);
+    }
 
-   private:
-    std::shared_ptr<ChannelInterface> channel_;
-    std::unique_ptr<TestService::Stub> stub_;
-  };
+    auto payload_size = config.payload_size();
+    auto check_done = [payload_size](grpc::Status s, SimpleResponse *response) {
+      GPR_ASSERT(s.IsOk() && (response->payload().type() ==
+                              grpc::testing::PayloadType::COMPRESSABLE) &&
+                 (response->payload().body().length() ==
+                  static_cast<size_t>(payload_size)));
+    };
+
+    int t = 0;
+    for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
+      for (auto &channel : channels_) {
+        auto *cq = cli_cqs_[t].get();
+        t = (t + 1) % cli_cqs_.size();
+        auto start_req = [cq](TestService::Stub *stub, grpc::ClientContext *ctx,
+                              const SimpleRequest &request, void *tag) {
+          return stub->AsyncUnaryCall(ctx, request, cq, tag);
+        };
+
+        TestService::Stub *stub = channel.get_stub();
+        const SimpleRequest &request = request_;
+        new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
+            stub, request, start_req, check_done);
+      }
+    }
 
-  std::vector<ClientChannelInfo> channels;
-  for (int i = 0; i < client_channels; i++) {
-    channels.push_back(ClientChannelInfo(oss.str()));
+    StartThreads(config.async_client_threads());
   }
 
-  std::vector<std::thread> threads;  // Will add threads when ready to execute
-  std::vector< ::gpr_histogram *> thread_stats(client_threads);
-
-  TestService::Stub *stub_stats = channels[0].get_stub();
-  grpc::ClientContext context_stats_begin;
-  StatsRequest stats_request;
-  ServerStats server_stats_begin;
-  stats_request.set_test_num(0);
-  grpc::Status status_beg = stub_stats->CollectServerStats(
-      &context_stats_begin, stats_request, &server_stats_begin);
-
-  grpc_profiler_start("qps_client_async.prof");
-
-  auto CheckDone = [=](grpc::Status s, SimpleResponse *response) {
-    GPR_ASSERT(s.IsOk() && (response->payload().type() ==
-                            grpc::testing::PayloadType::COMPRESSABLE) &&
-               (response->payload().body().length() ==
-                static_cast<size_t>(payload_size)));
-  };
+  ~AsyncClient() GRPC_OVERRIDE {
+    EndThreads();
 
-  for (int i = 0; i < client_threads; i++) {
-    gpr_histogram *hist = gpr_histogram_create(0.01, 60e9);
-    GPR_ASSERT(hist != NULL);
-    thread_stats[i] = hist;
-
-    threads.push_back(std::thread(
-        [hist, client_threads, client_channels, num_rpcs, payload_size,
-         &channels, &CheckDone](int channel_num) {
-          using namespace std::placeholders;
-          SimpleRequest request;
-          request.set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
-          request.set_response_size(payload_size);
-
-          grpc::CompletionQueue cli_cq;
-	  auto start_req = std::bind(&TestService::Stub::AsyncUnaryCall, _1,
-				     _2, _3, &cli_cq, _4);
-
-          int rpcs_sent = 0;
-          while (rpcs_sent < num_rpcs) {
-            rpcs_sent++;
-            TestService::Stub *stub = channels[channel_num].get_stub();
-            new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(stub,
-                request, start_req, CheckDone);
-            void *got_tag;
-            bool ok;
-
-            // Need to call 2 next for every 1 RPC (1 for req done, 1 for resp
-            // done)
-            cli_cq.Next(&got_tag, &ok);
-            if (!ok) break;
-            ClientRpcContext *ctx = ClientRpcContext::detag(got_tag);
-            if (ctx->RunNextState() == false) {
-              // call the callback and then delete it
-              ctx->report_stats(hist);
-              ctx->RunNextState();
-              delete ctx;
-            }
-            cli_cq.Next(&got_tag, &ok);
-            if (!ok) break;
-            ctx = ClientRpcContext::detag(got_tag);
-            if (ctx->RunNextState() == false) {
-              // call the callback and then delete it
-              ctx->report_stats(hist);
-	      ctx->RunNextState();
-              delete ctx;
-            }
-            // Now do runtime round-robin assignment of the next
-            // channel number
-            channel_num += client_threads;
-            channel_num %= client_channels;
-          }
-        },
-        i % client_channels));
-  }
-
-  gpr_histogram *hist = gpr_histogram_create(0.01, 60e9);
-  GPR_ASSERT(hist != NULL);
-  for (auto &t : threads) {
-    t.join();
+    for (auto &cq : cli_cqs_) {
+      cq->Shutdown();
+      void *got_tag;
+      bool ok;
+      while (cq->Next(&got_tag, &ok)) {
+        delete ClientRpcContext::detag(got_tag);
+      }
+    }
   }
 
-  grpc_profiler_stop();
-
-  for (int i = 0; i < client_threads; i++) {
-    gpr_histogram *h = thread_stats[i];
-    gpr_log(GPR_INFO, "latency at thread %d (50/90/95/99/99.9): %f/%f/%f/%f/%f",
-            i, gpr_histogram_percentile(h, 50), gpr_histogram_percentile(h, 90),
-            gpr_histogram_percentile(h, 95), gpr_histogram_percentile(h, 99),
-            gpr_histogram_percentile(h, 99.9));
-    gpr_histogram_merge(hist, h);
-    gpr_histogram_destroy(h);
+  void ThreadFunc(Histogram *histogram, size_t thread_idx) {
+    void *got_tag;
+    bool ok;
+    cli_cqs_[thread_idx]->Next(&got_tag, &ok);
+
+    ClientRpcContext *ctx = ClientRpcContext::detag(got_tag);
+    if (ctx->RunNextState() == false) {
+      // call the callback and then delete it
+      ctx->report_stats(histogram);
+      ctx->RunNextState();
+      ctx->StartNewClone();
+      delete ctx;
+    }
   }
 
-  gpr_log(
-      GPR_INFO,
-      "latency across %d threads with %d channels and %d payload "
-      "(50/90/95/99/99.9): %f / %f / %f / %f / %f",
-      client_threads, client_channels, payload_size,
-      gpr_histogram_percentile(hist, 50), gpr_histogram_percentile(hist, 90),
-      gpr_histogram_percentile(hist, 95), gpr_histogram_percentile(hist, 99),
-      gpr_histogram_percentile(hist, 99.9));
-  gpr_histogram_destroy(hist);
-
-  grpc::ClientContext context_stats_end;
-  ServerStats server_stats_end;
-  grpc::Status status_end = stub_stats->CollectServerStats(
-      &context_stats_end, stats_request, &server_stats_end);
+  std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
+};
 
-  double elapsed = server_stats_end.time_now() - server_stats_begin.time_now();
-  int total_rpcs = client_threads * num_rpcs;
-  double utime = server_stats_end.time_user() - server_stats_begin.time_user();
-  double stime =
-      server_stats_end.time_system() - server_stats_begin.time_system();
-  gpr_log(GPR_INFO,
-          "Elapsed time: %.3f\n"
-          "RPC Count: %d\n"
-          "QPS: %.3f\n"
-          "System time: %.3f\n"
-          "User time: %.3f\n"
-          "Resource usage: %.1f%%\n",
-          elapsed, total_rpcs, total_rpcs / elapsed, stime, utime,
-          (stime + utime) / elapsed * 100.0);
+std::unique_ptr<Client> CreateAsyncClient(const ClientConfig &args) {
+  return std::unique_ptr<Client>(new AsyncClient(args));
 }
 
-int main(int argc, char **argv) {
-  grpc_init();
-  ParseCommandLineFlags(&argc, &argv, true);
-
-  GPR_ASSERT(FLAGS_server_port);
-
-  if (FLAGS_workload.length() == 0) {
-    RunTest(FLAGS_client_threads, FLAGS_client_channels, FLAGS_num_rpcs,
-            FLAGS_payload_size);
-  } else {
-    std::istringstream workload(FLAGS_workload);
-    int client_threads, client_channels, num_rpcs, payload_size;
-    workload >> client_threads;
-    while (!workload.eof()) {
-      workload >> client_channels >> num_rpcs >> payload_size;
-      RunTest(client_threads, client_channels, num_rpcs, payload_size);
-      workload >> client_threads;
-    }
-    gpr_log(GPR_INFO, "Done with specified workload.");
-  }
-
-  grpc_shutdown();
-  return 0;
-}
+}  // namespace testing
+}  // namespace grpc