Added streaming C++ tests for sync and sync cases

1 files changed, 156 insertions, 17 deletions

diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index c6535bebf8..30317d11e1 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -46,6 +46,7 @@
 #include <grpc++/async_unary_call.h>
 #include <grpc++/client_context.h>
 #include <grpc++/status.h>
+#include <grpc++/stream.h>
 #include "test/core/util/grpc_profiler.h"
 #include "test/cpp/util/create_test_channel.h"
 #include "test/cpp/qps/qpstest.pb.h"
@@ -59,13 +60,13 @@ class ClientRpcContext {
  public:
   ClientRpcContext() {}
   virtual ~ClientRpcContext() {}
-  virtual bool RunNextState() = 0;  // do next state, return false if steps done
+  // next state, return false if done. Collect stats when appropriate
+  virtual bool RunNextState(bool, Histogram *hist) = 0;
   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(Histogram *hist) = 0;
 };
 
 template <class RequestType, class ResponseType>
@@ -89,9 +90,12 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
         response_reader_(
             start_req(stub_, &context_, req_, ClientRpcContext::tag(this))) {}
   ~ClientRpcContextUnaryImpl() GRPC_OVERRIDE {}
-  bool RunNextState() GRPC_OVERRIDE { return (this->*next_state_)(); }
-  void report_stats(Histogram *hist) GRPC_OVERRIDE {
-    hist->Add((Timer::Now() - start_) * 1e9);
+  bool RunNextState(bool ok, Histogram *hist) GRPC_OVERRIDE {
+    bool ret = (this->*next_state_)(ok);
+    if (!ret) {
+      hist->Add((Timer::Now() - start_) * 1e9);
+    }
+    return ret;
   }
 
   void StartNewClone() GRPC_OVERRIDE {
@@ -99,16 +103,16 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
   }
 
  private:
-  bool ReqSent() {
+  bool ReqSent(bool) {
     next_state_ = &ClientRpcContextUnaryImpl::RespDone;
     response_reader_->Finish(&response_, &status_, ClientRpcContext::tag(this));
     return true;
   }
-  bool RespDone() {
+  bool RespDone(bool) {
     next_state_ = &ClientRpcContextUnaryImpl::DoCallBack;
     return false;
   }
-  bool DoCallBack() {
+  bool DoCallBack(bool) {
     callback_(status_, &response_);
     return false;
   }
@@ -116,7 +120,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
   TestService::Stub *stub_;
   RequestType req_;
   ResponseType response_;
-  bool (ClientRpcContextUnaryImpl::*next_state_)();
+  bool (ClientRpcContextUnaryImpl::*next_state_)(bool);
   std::function<void(grpc::Status, ResponseType *)> callback_;
   std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
       TestService::Stub *, grpc::ClientContext *, const RequestType &, void *)>
@@ -127,9 +131,9 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
       response_reader_;
 };
 
-class AsyncClient GRPC_FINAL : public Client {
+class AsyncUnaryClient GRPC_FINAL : public Client {
  public:
-  explicit AsyncClient(const ClientConfig &config) : Client(config) {
+  explicit AsyncUnaryClient(const ClientConfig &config) : Client(config) {
     for (int i = 0; i < config.async_client_threads(); i++) {
       cli_cqs_.emplace_back(new CompletionQueue);
     }
@@ -162,7 +166,7 @@ class AsyncClient GRPC_FINAL : public Client {
     StartThreads(config.async_client_threads());
   }
 
-  ~AsyncClient() GRPC_OVERRIDE {
+  ~AsyncUnaryClient() GRPC_OVERRIDE {
     EndThreads();
 
     for (auto &cq : cli_cqs_) {
@@ -181,10 +185,9 @@ class AsyncClient GRPC_FINAL : public Client {
     cli_cqs_[thread_idx]->Next(&got_tag, &ok);
 
     ClientRpcContext *ctx = ClientRpcContext::detag(got_tag);
-    if (ctx->RunNextState() == false) {
+    if (ctx->RunNextState(ok, histogram) == false) {
       // call the callback and then delete it
-      ctx->report_stats(histogram);
-      ctx->RunNextState();
+      ctx->RunNextState(ok, histogram);
       ctx->StartNewClone();
       delete ctx;
     }
@@ -193,8 +196,144 @@ class AsyncClient GRPC_FINAL : public Client {
   std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
 };
 
-std::unique_ptr<Client> CreateAsyncClient(const ClientConfig &args) {
-  return std::unique_ptr<Client>(new AsyncClient(args));
+template <class RequestType, class ResponseType>
+class ClientRpcContextStreamingImpl : public ClientRpcContext {
+ public:
+  ClientRpcContextStreamingImpl(
+      TestService::Stub *stub, const RequestType &req,
+      std::function<
+              std::unique_ptr<grpc::ClientAsyncReaderWriter<
+                              RequestType,ResponseType>>(
+              TestService::Stub *, grpc::ClientContext *, void *)> start_req,
+      std::function<void(grpc::Status, ResponseType *)> on_done)
+      : context_(),
+        stub_(stub),
+        req_(req),
+        response_(),
+        next_state_(&ClientRpcContextStreamingImpl::ReqSent),
+        callback_(on_done),
+        start_req_(start_req),
+        start_(Timer::Now()),
+        stream_(start_req_(stub_, &context_, ClientRpcContext::tag(this))) {}
+  ~ClientRpcContextStreamingImpl() GRPC_OVERRIDE {}
+  bool RunNextState(bool ok, Histogram *hist) GRPC_OVERRIDE {
+    return (this->*next_state_)(ok, hist);
+  }
+  void StartNewClone() GRPC_OVERRIDE {
+    new ClientRpcContextStreamingImpl(stub_, req_, start_req_, callback_);
+  }
+
+ private:
+  bool ReqSent(bool ok, Histogram *) {
+    return StartWrite(ok);
+  }
+  bool StartWrite(bool ok) {
+    if (!ok) {
+      return(false);
+    }
+    start_ = Timer::Now();
+    next_state_ = &ClientRpcContextStreamingImpl::WriteDone;
+    stream_->Write(req_, ClientRpcContext::tag(this));
+    return true;
+  }
+  bool WriteDone(bool ok, Histogram *) {
+    if (!ok) {
+      return(false);
+    }
+    next_state_ = &ClientRpcContextStreamingImpl::ReadDone;
+    stream_->Read(&response_, ClientRpcContext::tag(this)); 
+    return true;
+  }
+  bool ReadDone(bool ok, Histogram *hist) {
+    hist->Add((Timer::Now() - start_) * 1e9);
+    return StartWrite(ok);
+  }
+  grpc::ClientContext context_;
+  TestService::Stub *stub_;
+  RequestType req_;
+  ResponseType response_;
+  bool (ClientRpcContextStreamingImpl::*next_state_)(bool, Histogram *);
+  std::function<void(grpc::Status, ResponseType *)> callback_;
+  std::function<std::unique_ptr<grpc::ClientAsyncReaderWriter<
+				  RequestType,ResponseType>>(
+      TestService::Stub *, grpc::ClientContext *, void *)> start_req_;
+  grpc::Status status_;
+  double start_;
+  std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType,ResponseType>>
+    stream_;
+};
+
+class AsyncStreamingClient GRPC_FINAL : public Client {
+ public:
+  explicit AsyncStreamingClient(const ClientConfig &config) : Client(config) {
+    for (int i = 0; i < config.async_client_threads(); i++) {
+      cli_cqs_.emplace_back(new CompletionQueue);
+    }
+
+    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,
+                              void *tag) {
+          auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
+	  return stream;
+        };
+
+        TestService::Stub *stub = channel.get_stub();
+        const SimpleRequest &request = request_;
+        new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
+            stub, request, start_req, check_done);
+      }
+    }
+
+    StartThreads(config.async_client_threads());
+  }
+
+  ~AsyncStreamingClient() GRPC_OVERRIDE {
+    EndThreads();
+
+    for (auto &cq : cli_cqs_) {
+      cq->Shutdown();
+      void *got_tag;
+      bool ok;
+      while (cq->Next(&got_tag, &ok)) {
+        delete ClientRpcContext::detag(got_tag);
+      }
+    }
+  }
+
+  void ThreadFunc(Histogram *histogram, size_t thread_idx) GRPC_OVERRIDE {
+    void *got_tag;
+    bool ok;
+    cli_cqs_[thread_idx]->Next(&got_tag, &ok);
+
+    ClientRpcContext *ctx = ClientRpcContext::detag(got_tag);
+    if (ctx->RunNextState(ok, histogram) == false) {
+      // call the callback and then delete it
+      ctx->RunNextState(ok, histogram);
+      ctx->StartNewClone();
+      delete ctx;
+    }
+  }
+
+  std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
+};
+
+std::unique_ptr<Client> CreateAsyncUnaryClient(const ClientConfig &args) {
+  return std::unique_ptr<Client>(new AsyncUnaryClient(args));
+}
+std::unique_ptr<Client> CreateAsyncStreamingClient(const ClientConfig &args) {
+  return std::unique_ptr<Client>(new AsyncStreamingClient(args));
 }
 
 }  // namespace testing