|
|
|
@ -61,23 +61,23 @@ class 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); |
|
|
|
|
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; |
|
|
|
|
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, |
|
|
|
|
std::function<void(grpc::Status, ResponseType *)> on_done) |
|
|
|
|
TestService::Stub*, grpc::ClientContext*, const RequestType&, |
|
|
|
|
void*)> start_req, |
|
|
|
|
std::function<void(grpc::Status, ResponseType*)> on_done) |
|
|
|
|
: context_(), |
|
|
|
|
stub_(stub), |
|
|
|
|
req_(req), |
|
|
|
@ -90,7 +90,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { |
|
|
|
|
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 { |
|
|
|
|
void report_stats(Histogram* hist) GRPC_OVERRIDE { |
|
|
|
|
hist->Add((Timer::Now() - start_) * 1e9); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -113,13 +113,13 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
grpc::ClientContext context_; |
|
|
|
|
TestService::Stub *stub_; |
|
|
|
|
TestService::Stub* stub_; |
|
|
|
|
RequestType req_; |
|
|
|
|
ResponseType response_; |
|
|
|
|
bool (ClientRpcContextUnaryImpl::*next_state_)(); |
|
|
|
|
std::function<void(grpc::Status, ResponseType *)> callback_; |
|
|
|
|
std::function<void(grpc::Status, ResponseType*)> callback_; |
|
|
|
|
std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( |
|
|
|
|
TestService::Stub *, grpc::ClientContext *, const RequestType &, void *)> |
|
|
|
|
TestService::Stub*, grpc::ClientContext*, const RequestType&, void*)> |
|
|
|
|
start_req_; |
|
|
|
|
grpc::Status status_; |
|
|
|
|
double start_; |
|
|
|
@ -129,13 +129,13 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { |
|
|
|
|
|
|
|
|
|
class AsyncClient GRPC_FINAL : public Client { |
|
|
|
|
public: |
|
|
|
|
explicit AsyncClient(const ClientConfig &config) : Client(config) { |
|
|
|
|
explicit AsyncClient(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) { |
|
|
|
|
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() == |
|
|
|
@ -144,16 +144,16 @@ class AsyncClient GRPC_FINAL : public Client { |
|
|
|
|
|
|
|
|
|
int t = 0; |
|
|
|
|
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) { |
|
|
|
|
for (auto &channel : channels_) { |
|
|
|
|
auto *cq = cli_cqs_[t].get(); |
|
|
|
|
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) { |
|
|
|
|
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_; |
|
|
|
|
TestService::Stub* stub = channel.get_stub(); |
|
|
|
|
const SimpleRequest& request = request_; |
|
|
|
|
new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( |
|
|
|
|
stub, request, start_req, check_done); |
|
|
|
|
} |
|
|
|
@ -165,9 +165,9 @@ class AsyncClient GRPC_FINAL : public Client { |
|
|
|
|
~AsyncClient() GRPC_OVERRIDE { |
|
|
|
|
EndThreads(); |
|
|
|
|
|
|
|
|
|
for (auto &cq : cli_cqs_) { |
|
|
|
|
for (auto& cq : cli_cqs_) { |
|
|
|
|
cq->Shutdown(); |
|
|
|
|
void *got_tag; |
|
|
|
|
void* got_tag; |
|
|
|
|
bool ok; |
|
|
|
|
while (cq->Next(&got_tag, &ok)) { |
|
|
|
|
delete ClientRpcContext::detag(got_tag); |
|
|
|
@ -175,12 +175,12 @@ class AsyncClient GRPC_FINAL : public Client { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void ThreadFunc(Histogram *histogram, size_t thread_idx) GRPC_OVERRIDE { |
|
|
|
|
void *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); |
|
|
|
|
ClientRpcContext* ctx = ClientRpcContext::detag(got_tag); |
|
|
|
|
if (ctx->RunNextState() == false) { |
|
|
|
|
// call the callback and then delete it
|
|
|
|
|
ctx->report_stats(histogram); |
|
|
|
@ -193,7 +193,7 @@ class AsyncClient GRPC_FINAL : public Client { |
|
|
|
|
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
std::unique_ptr<Client> CreateAsyncClient(const ClientConfig &args) { |
|
|
|
|
std::unique_ptr<Client> CreateAsyncClient(const ClientConfig& args) { |
|
|
|
|
return std::unique_ptr<Client>(new AsyncClient(args)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|