|
|
|
/*
|
|
|
|
*
|
|
|
|
* Copyright 2015-2016, Google Inc.
|
|
|
|
* All rights reserved.
|
|
|
|
*
|
|
|
|
* Redistribution and use in source and binary forms, with or without
|
|
|
|
* modification, are permitted provided that the following conditions are
|
|
|
|
* met:
|
|
|
|
*
|
|
|
|
* * Redistributions of source code must retain the above copyright
|
|
|
|
* notice, this list of conditions and the following disclaimer.
|
|
|
|
* * Redistributions in binary form must reproduce the above
|
|
|
|
* copyright notice, this list of conditions and the following disclaimer
|
|
|
|
* in the documentation and/or other materials provided with the
|
|
|
|
* distribution.
|
|
|
|
* * Neither the name of Google Inc. nor the names of its
|
|
|
|
* contributors may be used to endorse or promote products derived from
|
|
|
|
* this software without specific prior written permission.
|
|
|
|
*
|
|
|
|
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
|
|
|
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
|
|
|
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
|
|
|
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
|
|
|
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
|
|
|
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
|
|
|
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
|
|
|
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
|
|
|
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
|
|
|
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
|
|
|
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
|
|
|
*
|
|
|
|
*/
|
|
|
|
|
|
|
|
#include <cassert>
|
|
|
|
#include <forward_list>
|
|
|
|
#include <functional>
|
|
|
|
#include <list>
|
|
|
|
#include <memory>
|
|
|
|
#include <mutex>
|
|
|
|
#include <sstream>
|
|
|
|
#include <string>
|
|
|
|
#include <thread>
|
|
|
|
#include <vector>
|
|
|
|
|
|
|
|
#include <gflags/gflags.h>
|
|
|
|
#include <grpc++/alarm.h>
|
|
|
|
#include <grpc++/channel.h>
|
|
|
|
#include <grpc++/client_context.h>
|
|
|
|
#include <grpc++/generic/generic_stub.h>
|
|
|
|
#include <grpc/grpc.h>
|
|
|
|
#include <grpc/support/cpu.h>
|
|
|
|
#include <grpc/support/histogram.h>
|
|
|
|
#include <grpc/support/log.h>
|
|
|
|
|
|
|
|
#include "src/proto/grpc/testing/services.grpc.pb.h"
|
|
|
|
#include "test/cpp/qps/client.h"
|
|
|
|
#include "test/cpp/qps/usage_timer.h"
|
|
|
|
#include "test/cpp/util/create_test_channel.h"
|
|
|
|
|
|
|
|
namespace grpc {
|
|
|
|
namespace testing {
|
|
|
|
|
|
|
|
class ClientRpcContext {
|
|
|
|
public:
|
|
|
|
ClientRpcContext() {}
|
|
|
|
virtual ~ClientRpcContext() {}
|
|
|
|
// next state, return false if done. Collect stats when appropriate
|
|
|
|
virtual bool RunNextState(bool, Histogram* hist) = 0;
|
|
|
|
virtual ClientRpcContext* StartNewClone() = 0;
|
|
|
|
static void* tag(ClientRpcContext* c) { return reinterpret_cast<void*>(c); }
|
|
|
|
static ClientRpcContext* detag(void* t) {
|
|
|
|
return reinterpret_cast<ClientRpcContext*>(t);
|
|
|
|
}
|
|
|
|
|
|
|
|
virtual void Start(CompletionQueue* cq) = 0;
|
|
|
|
};
|
|
|
|
|
|
|
|
template <class RequestType, class ResponseType>
|
|
|
|
class ClientRpcContextUnaryImpl : public ClientRpcContext {
|
|
|
|
public:
|
|
|
|
ClientRpcContextUnaryImpl(
|
|
|
|
BenchmarkService::Stub* stub, const RequestType& req,
|
|
|
|
std::function<gpr_timespec()> next_issue,
|
|
|
|
std::function<
|
|
|
|
std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
|
|
|
|
BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
|
|
|
|
CompletionQueue*)> start_req,
|
|
|
|
std::function<void(grpc::Status, ResponseType*)> on_done)
|
|
|
|
: context_(),
|
|
|
|
stub_(stub),
|
|
|
|
cq_(nullptr),
|
|
|
|
req_(req),
|
|
|
|
response_(),
|
|
|
|
next_state_(State::READY),
|
|
|
|
callback_(on_done),
|
|
|
|
next_issue_(next_issue),
|
|
|
|
start_req_(start_req) {}
|
|
|
|
~ClientRpcContextUnaryImpl() GRPC_OVERRIDE {}
|
|
|
|
void Start(CompletionQueue* cq) GRPC_OVERRIDE {
|
|
|
|
cq_ = cq;
|
|
|
|
if (!next_issue_) { // ready to issue
|
|
|
|
RunNextState(true, nullptr);
|
|
|
|
} else { // wait for the issue time
|
|
|
|
alarm_.reset(new Alarm(cq_, next_issue_(), ClientRpcContext::tag(this)));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
|
|
|
|
switch (next_state_) {
|
|
|
|
case State::READY:
|
|
|
|
start_ = UsageTimer::Now();
|
|
|
|
response_reader_ = start_req_(stub_, &context_, req_, cq_);
|
|
|
|
response_reader_->Finish(&response_, &status_,
|
|
|
|
ClientRpcContext::tag(this));
|
|
|
|
next_state_ = State::RESP_DONE;
|
|
|
|
return true;
|
|
|
|
case State::RESP_DONE:
|
|
|
|
hist->Add((UsageTimer::Now() - start_) * 1e9);
|
|
|
|
callback_(status_, &response_);
|
|
|
|
next_state_ = State::INVALID;
|
|
|
|
return false;
|
|
|
|
default:
|
|
|
|
GPR_ASSERT(false);
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
|
|
|
|
return new ClientRpcContextUnaryImpl(stub_, req_, next_issue_, start_req_,
|
|
|
|
callback_);
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
grpc::ClientContext context_;
|
|
|
|
BenchmarkService::Stub* stub_;
|
|
|
|
CompletionQueue* cq_;
|
|
|
|
std::unique_ptr<Alarm> alarm_;
|
|
|
|
RequestType req_;
|
|
|
|
ResponseType response_;
|
|
|
|
enum State { INVALID, READY, RESP_DONE };
|
|
|
|
State next_state_;
|
|
|
|
std::function<void(grpc::Status, ResponseType*)> callback_;
|
|
|
|
std::function<gpr_timespec()> next_issue_;
|
|
|
|
std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
|
|
|
|
BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
|
|
|
|
CompletionQueue*)> start_req_;
|
|
|
|
grpc::Status status_;
|
|
|
|
double start_;
|
|
|
|
std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>
|
|
|
|
response_reader_;
|
|
|
|
};
|
|
|
|
|
|
|
|
typedef std::forward_list<ClientRpcContext*> context_list;
|
|
|
|
|
|
|
|
template <class StubType, class RequestType>
|
|
|
|
class AsyncClient : public ClientImpl<StubType, RequestType> {
|
|
|
|
// Specify which protected members we are using since there is no
|
|
|
|
// member name resolution until the template types are fully resolved
|
|
|
|
public:
|
|
|
|
using Client::SetupLoadTest;
|
|
|
|
using Client::closed_loop_;
|
|
|
|
using Client::NextIssuer;
|
|
|
|
using ClientImpl<StubType, RequestType>::cores_;
|
|
|
|
using ClientImpl<StubType, RequestType>::channels_;
|
|
|
|
using ClientImpl<StubType, RequestType>::request_;
|
|
|
|
AsyncClient(const ClientConfig& config,
|
|
|
|
std::function<ClientRpcContext*(
|
|
|
|
StubType*, std::function<gpr_timespec()> next_issue,
|
|
|
|
const RequestType&)> setup_ctx,
|
|
|
|
std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)>
|
|
|
|
create_stub)
|
|
|
|
: ClientImpl<StubType, RequestType>(config, create_stub),
|
|
|
|
num_async_threads_(NumThreads(config)) {
|
|
|
|
SetupLoadTest(config, num_async_threads_);
|
|
|
|
|
|
|
|
for (int i = 0; i < num_async_threads_; i++) {
|
|
|
|
cli_cqs_.emplace_back(new CompletionQueue);
|
|
|
|
next_issuers_.emplace_back(NextIssuer(i));
|
|
|
|
}
|
|
|
|
|
|
|
|
using namespace std::placeholders;
|
|
|
|
int t = 0;
|
|
|
|
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
|
|
|
|
for (int ch = 0; ch < config.client_channels(); ch++) {
|
|
|
|
auto* cq = cli_cqs_[t].get();
|
|
|
|
auto ctx =
|
|
|
|
setup_ctx(channels_[ch].get_stub(), next_issuers_[t], request_);
|
|
|
|
ctx->Start(cq);
|
|
|
|
t = (t + 1) % cli_cqs_.size();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
virtual ~AsyncClient() {
|
|
|
|
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);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
bool ThreadFunc(Histogram* histogram,
|
|
|
|
size_t thread_idx) GRPC_OVERRIDE GRPC_FINAL {
|
|
|
|
void* got_tag;
|
|
|
|
bool ok;
|
|
|
|
|
|
|
|
if (cli_cqs_[thread_idx]->Next(&got_tag, &ok)) {
|
|
|
|
// Got a regular event, so process it
|
|
|
|
ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
|
|
|
|
if (!ctx->RunNextState(ok, histogram)) {
|
|
|
|
// The RPC and callback are done, so clone the ctx
|
|
|
|
// and kickstart the new one
|
|
|
|
auto clone = ctx->StartNewClone();
|
|
|
|
clone->Start(cli_cqs_[thread_idx].get());
|
|
|
|
// delete the old version
|
|
|
|
delete ctx;
|
|
|
|
}
|
|
|
|
return true;
|
|
|
|
} else { // queue is shutting down
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
protected:
|
|
|
|
const int num_async_threads_;
|
|
|
|
|
|
|
|
private:
|
|
|
|
int NumThreads(const ClientConfig& config) {
|
|
|
|
int num_threads = config.async_client_threads();
|
|
|
|
if (num_threads <= 0) { // Use dynamic sizing
|
|
|
|
num_threads = cores_;
|
|
|
|
gpr_log(GPR_INFO, "Sizing async client to %d threads", num_threads);
|
|
|
|
}
|
|
|
|
return num_threads;
|
|
|
|
}
|
|
|
|
|
|
|
|
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
|
|
|
|
std::vector<std::function<gpr_timespec()>> next_issuers_;
|
|
|
|
};
|
|
|
|
|
|
|
|
static std::unique_ptr<BenchmarkService::Stub> BenchmarkStubCreator(
|
|
|
|
std::shared_ptr<Channel> ch) {
|
|
|
|
return BenchmarkService::NewStub(ch);
|
|
|
|
}
|
|
|
|
|
|
|
|
class AsyncUnaryClient GRPC_FINAL
|
|
|
|
: public AsyncClient<BenchmarkService::Stub, SimpleRequest> {
|
|
|
|
public:
|
|
|
|
explicit AsyncUnaryClient(const ClientConfig& config)
|
|
|
|
: AsyncClient(config, SetupCtx, BenchmarkStubCreator) {
|
|
|
|
StartThreads(num_async_threads_);
|
|
|
|
}
|
|
|
|
~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); }
|
|
|
|
|
|
|
|
private:
|
|
|
|
static void CheckDone(grpc::Status s, SimpleResponse* response) {}
|
|
|
|
static std::unique_ptr<grpc::ClientAsyncResponseReader<SimpleResponse>>
|
|
|
|
StartReq(BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
|
|
|
|
const SimpleRequest& request, CompletionQueue* cq) {
|
|
|
|
return stub->AsyncUnaryCall(ctx, request, cq);
|
|
|
|
};
|
|
|
|
static ClientRpcContext* SetupCtx(BenchmarkService::Stub* stub,
|
|
|
|
std::function<gpr_timespec()> next_issue,
|
|
|
|
const SimpleRequest& req) {
|
|
|
|
return new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
|
|
|
|
stub, req, next_issue, AsyncUnaryClient::StartReq,
|
|
|
|
AsyncUnaryClient::CheckDone);
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
template <class RequestType, class ResponseType>
|
|
|
|
class ClientRpcContextStreamingImpl : public ClientRpcContext {
|
|
|
|
public:
|
|
|
|
ClientRpcContextStreamingImpl(
|
|
|
|
BenchmarkService::Stub* stub, const RequestType& req,
|
|
|
|
std::function<gpr_timespec()> next_issue,
|
|
|
|
std::function<std::unique_ptr<
|
|
|
|
grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
|
|
|
|
BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*,
|
|
|
|
void*)> start_req,
|
|
|
|
std::function<void(grpc::Status, ResponseType*)> on_done)
|
|
|
|
: context_(),
|
|
|
|
stub_(stub),
|
|
|
|
cq_(nullptr),
|
|
|
|
req_(req),
|
|
|
|
response_(),
|
|
|
|
next_state_(State::INVALID),
|
|
|
|
callback_(on_done),
|
|
|
|
next_issue_(next_issue),
|
|
|
|
start_req_(start_req) {}
|
|
|
|
~ClientRpcContextStreamingImpl() GRPC_OVERRIDE {}
|
|
|
|
void Start(CompletionQueue* cq) GRPC_OVERRIDE {
|
|
|
|
cq_ = cq;
|
|
|
|
stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this));
|
|
|
|
next_state_ = State::STREAM_IDLE;
|
|
|
|
}
|
|
|
|
bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
|
|
|
|
while (true) {
|
|
|
|
switch (next_state_) {
|
|
|
|
case State::STREAM_IDLE:
|
|
|
|
if (!next_issue_) { // ready to issue
|
|
|
|
next_state_ = State::READY_TO_WRITE;
|
|
|
|
} else {
|
|
|
|
next_state_ = State::WAIT;
|
|
|
|
}
|
|
|
|
break; // loop around, don't return
|
|
|
|
case State::WAIT:
|
|
|
|
alarm_.reset(
|
|
|
|
new Alarm(cq_, next_issue_(), ClientRpcContext::tag(this)));
|
|
|
|
next_state_ = State::READY_TO_WRITE;
|
|
|
|
return true;
|
|
|
|
case State::READY_TO_WRITE:
|
|
|
|
if (!ok) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
start_ = UsageTimer::Now();
|
|
|
|
next_state_ = State::WRITE_DONE;
|
|
|
|
stream_->Write(req_, ClientRpcContext::tag(this));
|
|
|
|
return true;
|
|
|
|
case State::WRITE_DONE:
|
|
|
|
if (!ok) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
next_state_ = State::READ_DONE;
|
|
|
|
stream_->Read(&response_, ClientRpcContext::tag(this));
|
|
|
|
return true;
|
|
|
|
break;
|
|
|
|
case State::READ_DONE:
|
|
|
|
hist->Add((UsageTimer::Now() - start_) * 1e9);
|
|
|
|
callback_(status_, &response_);
|
|
|
|
next_state_ = State::STREAM_IDLE;
|
|
|
|
break; // loop around
|
|
|
|
default:
|
|
|
|
GPR_ASSERT(false);
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
|
|
|
|
return new ClientRpcContextStreamingImpl(stub_, req_, next_issue_,
|
|
|
|
start_req_, callback_);
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
grpc::ClientContext context_;
|
|
|
|
BenchmarkService::Stub* stub_;
|
|
|
|
CompletionQueue* cq_;
|
|
|
|
std::unique_ptr<Alarm> alarm_;
|
|
|
|
RequestType req_;
|
|
|
|
ResponseType response_;
|
|
|
|
enum State {
|
|
|
|
INVALID,
|
|
|
|
STREAM_IDLE,
|
|
|
|
WAIT,
|
|
|
|
READY_TO_WRITE,
|
|
|
|
WRITE_DONE,
|
|
|
|
READ_DONE
|
|
|
|
};
|
|
|
|
State next_state_;
|
|
|
|
std::function<void(grpc::Status, ResponseType*)> callback_;
|
|
|
|
std::function<gpr_timespec()> next_issue_;
|
|
|
|
std::function<
|
|
|
|
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
|
|
|
|
BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*,
|
|
|
|
void*)> start_req_;
|
|
|
|
grpc::Status status_;
|
|
|
|
double start_;
|
|
|
|
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>
|
|
|
|
stream_;
|
|
|
|
};
|
|
|
|
|
|
|
|
class AsyncStreamingClient GRPC_FINAL
|
|
|
|
: public AsyncClient<BenchmarkService::Stub, SimpleRequest> {
|
|
|
|
public:
|
|
|
|
explicit AsyncStreamingClient(const ClientConfig& config)
|
|
|
|
: AsyncClient(config, SetupCtx, BenchmarkStubCreator) {
|
|
|
|
StartThreads(num_async_threads_);
|
|
|
|
}
|
|
|
|
|
|
|
|
~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
|
|
|
|
|
|
|
|
private:
|
|
|
|
static void CheckDone(grpc::Status s, SimpleResponse* response) {}
|
|
|
|
static std::unique_ptr<
|
|
|
|
grpc::ClientAsyncReaderWriter<SimpleRequest, SimpleResponse>>
|
|
|
|
StartReq(BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
|
|
|
|
CompletionQueue* cq, void* tag) {
|
|
|
|
auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
|
|
|
|
return stream;
|
|
|
|
};
|
|
|
|
static ClientRpcContext* SetupCtx(BenchmarkService::Stub* stub,
|
|
|
|
std::function<gpr_timespec()> next_issue,
|
|
|
|
const SimpleRequest& req) {
|
|
|
|
return new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
|
|
|
|
stub, req, next_issue, AsyncStreamingClient::StartReq,
|
|
|
|
AsyncStreamingClient::CheckDone);
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
|
|
|
|
public:
|
|
|
|
ClientRpcContextGenericStreamingImpl(
|
|
|
|
grpc::GenericStub* stub, const ByteBuffer& req,
|
|
|
|
std::function<gpr_timespec()> next_issue,
|
|
|
|
std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>(
|
|
|
|
grpc::GenericStub*, grpc::ClientContext*,
|
|
|
|
const grpc::string& method_name, CompletionQueue*, void*)> start_req,
|
|
|
|
std::function<void(grpc::Status, ByteBuffer*)> on_done)
|
|
|
|
: context_(),
|
|
|
|
stub_(stub),
|
|
|
|
cq_(nullptr),
|
|
|
|
req_(req),
|
|
|
|
response_(),
|
|
|
|
next_state_(State::INVALID),
|
|
|
|
callback_(on_done),
|
|
|
|
next_issue_(next_issue),
|
|
|
|
start_req_(start_req) {}
|
|
|
|
~ClientRpcContextGenericStreamingImpl() GRPC_OVERRIDE {}
|
|
|
|
void Start(CompletionQueue* cq) GRPC_OVERRIDE {
|
|
|
|
cq_ = cq;
|
|
|
|
const grpc::string kMethodName(
|
|
|
|
"/grpc.testing.BenchmarkService/StreamingCall");
|
|
|
|
stream_ = start_req_(stub_, &context_, kMethodName, cq,
|
|
|
|
ClientRpcContext::tag(this));
|
|
|
|
next_state_ = State::STREAM_IDLE;
|
|
|
|
}
|
|
|
|
bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
|
|
|
|
while (true) {
|
|
|
|
switch (next_state_) {
|
|
|
|
case State::STREAM_IDLE:
|
|
|
|
if (!next_issue_) { // ready to issue
|
|
|
|
next_state_ = State::READY_TO_WRITE;
|
|
|
|
} else {
|
|
|
|
next_state_ = State::WAIT;
|
|
|
|
}
|
|
|
|
break; // loop around, don't return
|
|
|
|
case State::WAIT:
|
|
|
|
alarm_.reset(
|
|
|
|
new Alarm(cq_, next_issue_(), ClientRpcContext::tag(this)));
|
|
|
|
next_state_ = State::READY_TO_WRITE;
|
|
|
|
return true;
|
|
|
|
case State::READY_TO_WRITE:
|
|
|
|
if (!ok) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
start_ = UsageTimer::Now();
|
|
|
|
next_state_ = State::WRITE_DONE;
|
|
|
|
stream_->Write(req_, ClientRpcContext::tag(this));
|
|
|
|
return true;
|
|
|
|
case State::WRITE_DONE:
|
|
|
|
if (!ok) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
next_state_ = State::READ_DONE;
|
|
|
|
stream_->Read(&response_, ClientRpcContext::tag(this));
|
|
|
|
return true;
|
|
|
|
break;
|
|
|
|
case State::READ_DONE:
|
|
|
|
hist->Add((UsageTimer::Now() - start_) * 1e9);
|
|
|
|
callback_(status_, &response_);
|
|
|
|
next_state_ = State::STREAM_IDLE;
|
|
|
|
break; // loop around
|
|
|
|
default:
|
|
|
|
GPR_ASSERT(false);
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
|
|
|
|
return new ClientRpcContextGenericStreamingImpl(stub_, req_, next_issue_,
|
|
|
|
start_req_, callback_);
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
grpc::ClientContext context_;
|
|
|
|
grpc::GenericStub* stub_;
|
|
|
|
CompletionQueue* cq_;
|
|
|
|
std::unique_ptr<Alarm> alarm_;
|
|
|
|
ByteBuffer req_;
|
|
|
|
ByteBuffer response_;
|
|
|
|
enum State {
|
|
|
|
INVALID,
|
|
|
|
STREAM_IDLE,
|
|
|
|
WAIT,
|
|
|
|
READY_TO_WRITE,
|
|
|
|
WRITE_DONE,
|
|
|
|
READ_DONE
|
|
|
|
};
|
|
|
|
State next_state_;
|
|
|
|
std::function<void(grpc::Status, ByteBuffer*)> callback_;
|
|
|
|
std::function<gpr_timespec()> next_issue_;
|
|
|
|
std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>(
|
|
|
|
grpc::GenericStub*, grpc::ClientContext*, const grpc::string&,
|
|
|
|
CompletionQueue*, void*)> start_req_;
|
|
|
|
grpc::Status status_;
|
|
|
|
double start_;
|
|
|
|
std::unique_ptr<grpc::GenericClientAsyncReaderWriter> stream_;
|
|
|
|
};
|
|
|
|
|
|
|
|
static std::unique_ptr<grpc::GenericStub> GenericStubCreator(
|
|
|
|
std::shared_ptr<Channel> ch) {
|
|
|
|
return std::unique_ptr<grpc::GenericStub>(new grpc::GenericStub(ch));
|
|
|
|
}
|
|
|
|
|
|
|
|
class GenericAsyncStreamingClient GRPC_FINAL
|
|
|
|
: public AsyncClient<grpc::GenericStub, ByteBuffer> {
|
|
|
|
public:
|
|
|
|
explicit GenericAsyncStreamingClient(const ClientConfig& config)
|
|
|
|
: AsyncClient(config, SetupCtx, GenericStubCreator) {
|
|
|
|
StartThreads(num_async_threads_);
|
|
|
|
}
|
|
|
|
|
|
|
|
~GenericAsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
|
|
|
|
|
|
|
|
private:
|
|
|
|
static void CheckDone(grpc::Status s, ByteBuffer* response) {}
|
|
|
|
static std::unique_ptr<grpc::GenericClientAsyncReaderWriter> StartReq(
|
|
|
|
grpc::GenericStub* stub, grpc::ClientContext* ctx,
|
|
|
|
const grpc::string& method_name, CompletionQueue* cq, void* tag) {
|
|
|
|
auto stream = stub->Call(ctx, method_name, cq, tag);
|
|
|
|
return stream;
|
|
|
|
};
|
|
|
|
static ClientRpcContext* SetupCtx(grpc::GenericStub* stub,
|
|
|
|
std::function<gpr_timespec()> next_issue,
|
|
|
|
const ByteBuffer& req) {
|
|
|
|
return new ClientRpcContextGenericStreamingImpl(
|
|
|
|
stub, req, next_issue, GenericAsyncStreamingClient::StartReq,
|
|
|
|
GenericAsyncStreamingClient::CheckDone);
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
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));
|
|
|
|
}
|
|
|
|
std::unique_ptr<Client> CreateGenericAsyncStreamingClient(
|
|
|
|
const ClientConfig& args) {
|
|
|
|
return std::unique_ptr<Client>(new GenericAsyncStreamingClient(args));
|
|
|
|
}
|
|
|
|
|
|
|
|
} // namespace testing
|
|
|
|
} // namespace grpc
|