aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp/qps/client_async.cc
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp/qps/client_async.cc')
-rw-r--r--test/cpp/qps/client_async.cc185
1 files changed, 163 insertions, 22 deletions
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 526f37a1fd..590d56d8d0 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);
}
@@ -144,7 +148,8 @@ class AsyncClient GRPC_FINAL : public Client {
int t = 0;
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
- for (auto& channel : channels_) {
+ for (auto channel = channels_.begin(); channel != channels_.end();
+ channel++) {
auto* cq = cli_cqs_[t].get();
t = (t + 1) % cli_cqs_.size();
auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx,
@@ -152,7 +157,7 @@ class AsyncClient GRPC_FINAL : public Client {
return stub->AsyncUnaryCall(ctx, request, cq, tag);
};
- TestService::Stub* stub = channel.get_stub();
+ TestService::Stub* stub = channel->get_stub();
const SimpleRequest& request = request_;
new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
stub, request, start_req, check_done);
@@ -162,14 +167,14 @@ class AsyncClient GRPC_FINAL : public Client {
StartThreads(config.async_client_threads());
}
- ~AsyncClient() GRPC_OVERRIDE {
+ ~AsyncUnaryClient() GRPC_OVERRIDE {
EndThreads();
- for (auto& cq : cli_cqs_) {
- cq->Shutdown();
+ for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) {
+ (*cq)->Shutdown();
void* got_tag;
bool ok;
- while (cq->Next(&got_tag, &ok)) {
+ while ((*cq)->Next(&got_tag, &ok)) {
delete ClientRpcContext::detag(got_tag);
}
}
@@ -181,10 +186,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 +197,145 @@ 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_.begin(); channel != channels_.end();
+ channel++) {
+ 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_.begin(); cq != cli_cqs_.end(); cq++) {
+ (*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
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-03 20:53:39 +0000