Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

clang-format

Browse Source
pull/1948/head
Vijay Pai 10 years ago
parent c7de81e57e
commit cf3fb092cc
8 changed files with 260 additions and 247 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 22
      include/grpc++/config.h
  2. 18
      include/grpc++/time.h
  3. 2
      src/cpp/util/time.cc
  4. 47
      test/cpp/qps/client.h
  5. 287
      test/cpp/qps/client_async.cc
  6. 90
      test/cpp/qps/interarrival.h
  7. 14
      test/cpp/qps/qps_interarrival_test.cc
  8. 19
      test/cpp/qps/qpstest.proto

22
include/grpc++/config.h
Unescape View File

@ -99,24 +99,28 @@
::google::protobuf::io::ZeroCopyOutputStream ::google::protobuf::io::ZeroCopyOutputStream
#define GRPC_CUSTOM_ZEROCOPYINPUTSTREAM \ #define GRPC_CUSTOM_ZEROCOPYINPUTSTREAM \
::google::protobuf::io::ZeroCopyInputStream ::google::protobuf::io::ZeroCopyInputStream
#define GRPC_CUSTOM_CODEDINPUTSTREAM \ #define GRPC_CUSTOM_CODEDINPUTSTREAM ::google::protobuf::io::CodedInputStream
::google::protobuf::io::CodedInputStream
#endif #endif
#ifdef GRPC_CXX0X_NO_NULLPTR #ifdef GRPC_CXX0X_NO_NULLPTR
#include <memory> #include <memory>
const class { const class {
public: public:
template <class T> operator T*() const {return static_cast<T *>(0);} template <class T>
template <class T> operator std::unique_ptr<T>() const { operator T *() const {
return static_cast<T *>(0);
}
template <class T>
operator std::unique_ptr<T>() const {
return std::unique_ptr<T>(static_cast<T *>(0)); return std::unique_ptr<T>(static_cast<T *>(0));
} }
template <class T> operator std::shared_ptr<T>() const { template <class T>
operator std::shared_ptr<T>() const {
return std::shared_ptr<T>(static_cast<T *>(0)); return std::shared_ptr<T>(static_cast<T *>(0));
} }
operator bool() const {return false;} operator bool() const { return false; }
private:
private:
void operator&() const = delete; void operator&() const = delete;
} nullptr = {}; } nullptr = {};
#endif #endif

18
include/grpc++/time.h
Unescape View File

@ -52,22 +52,22 @@ namespace grpc {
template <typename T> template <typename T>
class TimePoint { class TimePoint {
public: public:
TimePoint(const T& time) { TimePoint(const T& time) { you_need_a_specialization_of_TimePoint(); }
you_need_a_specialization_of_TimePoint();
}
gpr_timespec raw_time() { gpr_timespec raw_time() {
gpr_timespec t; gpr_timespec t;
return t; return t;
} }
private: private:
void you_need_a_specialization_of_TimePoint(); void you_need_a_specialization_of_TimePoint();
}; };
template<> template <>
class TimePoint<gpr_timespec> { class TimePoint<gpr_timespec> {
public: public:
TimePoint(const gpr_timespec& time) : time_(time) { } TimePoint(const gpr_timespec& time) : time_(time) {}
gpr_timespec raw_time() { return time_; } gpr_timespec raw_time() { return time_; }
private: private:
gpr_timespec time_; gpr_timespec time_;
}; };
@ -85,8 +85,9 @@ namespace grpc {
// from and to should be absolute time. // from and to should be absolute time.
void Timepoint2Timespec(const std::chrono::system_clock::time_point& from, void Timepoint2Timespec(const std::chrono::system_clock::time_point& from,
gpr_timespec* to); gpr_timespec* to);
void TimepointHR2Timespec(const std::chrono::high_resolution_clock::time_point& from, void TimepointHR2Timespec(
gpr_timespec* to); const std::chrono::high_resolution_clock::time_point& from,
gpr_timespec* to);
std::chrono::system_clock::time_point Timespec2Timepoint(gpr_timespec t); std::chrono::system_clock::time_point Timespec2Timepoint(gpr_timespec t);
@ -94,9 +95,10 @@ template <>
class TimePoint<std::chrono::system_clock::time_point> { class TimePoint<std::chrono::system_clock::time_point> {
public: public:
TimePoint(const std::chrono::system_clock::time_point& time) { TimePoint(const std::chrono::system_clock::time_point& time) {
Timepoint2Timespec(time, &time_); Timepoint2Timespec(time, &time_);
} }
gpr_timespec raw_time() const { return time_; } gpr_timespec raw_time() const { return time_; }
private: private:
gpr_timespec time_; gpr_timespec time_;
}; };

2
src/cpp/util/time.cc
Unescape View File

@ -61,7 +61,7 @@ void Timepoint2Timespec(const system_clock::time_point& from,
} }
void TimepointHR2Timespec(const high_resolution_clock::time_point& from, void TimepointHR2Timespec(const high_resolution_clock::time_point& from,
gpr_timespec* to) { gpr_timespec* to) {
high_resolution_clock::duration deadline = from.time_since_epoch(); high_resolution_clock::duration deadline = from.time_since_epoch();
seconds secs = duration_cast<seconds>(deadline); seconds secs = duration_cast<seconds>(deadline);
if (from == high_resolution_clock::time_point::max() || if (from == high_resolution_clock::time_point::max() ||

47
test/cpp/qps/client.h
Unescape View File

@ -50,9 +50,10 @@ template <>
class TimePoint<std::chrono::high_resolution_clock::time_point> { class TimePoint<std::chrono::high_resolution_clock::time_point> {
public: public:
TimePoint(const std::chrono::high_resolution_clock::time_point& time) { TimePoint(const std::chrono::high_resolution_clock::time_point& time) {
TimepointHR2Timespec(time, &time_); TimepointHR2Timespec(time, &time_);
} }
gpr_timespec raw_time() const { return time_; } gpr_timespec raw_time() const { return time_; }
private: private:
gpr_timespec time_; gpr_timespec time_;
}; };
@ -65,8 +66,8 @@ typedef std::chrono::time_point<grpc_time_source> grpc_time;
class Client { class Client {
public: public:
explicit Client(const ClientConfig& config) : timer_(new Timer), explicit Client(const ClientConfig& config)
interarrival_timer_() { : timer_(new Timer), interarrival_timer_() {
for (int i = 0; i < config.client_channels(); i++) { for (int i = 0; i < config.client_channels(); i++) {
channels_.push_back(ClientChannelInfo( channels_.push_back(ClientChannelInfo(
config.server_targets(i % config.server_targets_size()), config)); config.server_targets(i % config.server_targets_size()), config));
@ -131,30 +132,29 @@ class Client {
// Set up the load distribution based on the number of threads // Set up the load distribution based on the number of threads
if (config.load_type() == CLOSED_LOOP) { if (config.load_type() == CLOSED_LOOP) {
closed_loop_ = true; closed_loop_ = true;
} } else {
else {
closed_loop_ = false; closed_loop_ = false;
std::unique_ptr<RandomDist> random_dist; std::unique_ptr<RandomDist> random_dist;
auto& load = config.load_params(); auto& load = config.load_params();
switch (config.load_type()) { switch (config.load_type()) {
case POISSON: case POISSON:
random_dist.reset random_dist.reset(
(new ExpDist(load.poisson().offered_load()/num_threads)); new ExpDist(load.poisson().offered_load() / num_threads));
break; break;
case UNIFORM: case UNIFORM:
random_dist.reset random_dist.reset(
(new UniformDist(load.uniform().interarrival_lo()*num_threads, new UniformDist(load.uniform().interarrival_lo() * num_threads,
load.uniform().interarrival_hi()*num_threads)); load.uniform().interarrival_hi() * num_threads));
break; break;
case DETERMINISTIC: case DETERMINISTIC:
random_dist.reset random_dist.reset(
(new DetDist(num_threads/load.determ().offered_load())); new DetDist(num_threads / load.determ().offered_load()));
break; break;
case PARETO: case PARETO:
random_dist.reset random_dist.reset(
(new ParetoDist(load.pareto().interarrival_base()*num_threads, new ParetoDist(load.pareto().interarrival_base() * num_threads,
load.pareto().alpha())); load.pareto().alpha()));
break; break;
default: default:
GPR_ASSERT(false); GPR_ASSERT(false);
@ -162,19 +162,22 @@ class Client {
} }
interarrival_timer_.init(*random_dist, num_threads); interarrival_timer_.init(*random_dist, num_threads);
for (size_t i = 0; i<num_threads; i++) { for (size_t i = 0; i < num_threads; i++) {
next_time_.push_back(grpc_time_source::now() + next_time_.push_back(
std::chrono::duration_cast<grpc_time_source::duration>(interarrival_timer_(i))); grpc_time_source::now() +
std::chrono::duration_cast<grpc_time_source::duration>(
interarrival_timer_(i)));
} }
} }
} }
bool NextIssueTime(int thread_idx, grpc_time *time_delay) { bool NextIssueTime(int thread_idx, grpc_time* time_delay) {
if (closed_loop_) { if (closed_loop_) {
return false; return false;
} } else {
else {
*time_delay = next_time_[thread_idx]; *time_delay = next_time_[thread_idx];
next_time_[thread_idx] += std::chrono::duration_cast<grpc_time_source::duration>(interarrival_timer_(thread_idx)); next_time_[thread_idx] +=
std::chrono::duration_cast<grpc_time_source::duration>(
interarrival_timer_(thread_idx));
return true; return true;
} }
} }

287
test/cpp/qps/client_async.cc
Unescape View File

@ -62,7 +62,7 @@ typedef std::list<grpc_time> deadline_list;
class ClientRpcContext { class ClientRpcContext {
public: public:
ClientRpcContext(int ch): channel_id_(ch) {} ClientRpcContext(int ch) : channel_id_(ch) {}
virtual ~ClientRpcContext() {} virtual ~ClientRpcContext() {}
// next state, return false if done. Collect stats when appropriate // next state, return false if done. Collect stats when appropriate
virtual bool RunNextState(bool, Histogram* hist) = 0; virtual bool RunNextState(bool, Histogram* hist) = 0;
@ -72,12 +72,16 @@ class ClientRpcContext {
return reinterpret_cast<ClientRpcContext*>(t); return reinterpret_cast<ClientRpcContext*>(t);
} }
deadline_list::iterator deadline_posn() const {return deadline_posn_;} deadline_list::iterator deadline_posn() const { return deadline_posn_; }
void set_deadline_posn(const deadline_list::iterator& it) {deadline_posn_ = it;} void set_deadline_posn(const deadline_list::iterator& it) {
virtual void Start(CompletionQueue *cq) = 0; deadline_posn_ = it;
int channel_id() const {return channel_id_;} }
virtual void Start(CompletionQueue* cq) = 0;
int channel_id() const { return channel_id_; }
protected: protected:
int channel_id_; int channel_id_;
private: private:
deadline_list::iterator deadline_posn_; deadline_list::iterator deadline_posn_;
}; };
@ -85,23 +89,22 @@ class ClientRpcContext {
template <class RequestType, class ResponseType> template <class RequestType, class ResponseType>
class ClientRpcContextUnaryImpl : public ClientRpcContext { class ClientRpcContextUnaryImpl : public ClientRpcContext {
public: public:
ClientRpcContextUnaryImpl(int channel_id, ClientRpcContextUnaryImpl(
TestService::Stub* stub, const RequestType& req, int channel_id, TestService::Stub* stub, const RequestType& req,
std::function< std::function<
std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
TestService::Stub*, grpc::ClientContext*, const RequestType&, TestService::Stub*, grpc::ClientContext*, const RequestType&,
CompletionQueue*)> CompletionQueue*)> start_req,
start_req,
std::function<void(grpc::Status, ResponseType*)> on_done) std::function<void(grpc::Status, ResponseType*)> on_done)
: ClientRpcContext(channel_id), context_(), : ClientRpcContext(channel_id),
context_(),
stub_(stub), stub_(stub),
req_(req), req_(req),
response_(), response_(),
next_state_(&ClientRpcContextUnaryImpl::RespDone), next_state_(&ClientRpcContextUnaryImpl::RespDone),
callback_(on_done), callback_(on_done),
start_req_(start_req) { start_req_(start_req) {}
} void Start(CompletionQueue* cq) GRPC_OVERRIDE {
void Start(CompletionQueue *cq) GRPC_OVERRIDE {
start_ = Timer::Now(); start_ = Timer::Now();
response_reader_ = start_req_(stub_, &context_, req_, cq); response_reader_ = start_req_(stub_, &context_, req_, cq);
response_reader_->Finish(&response_, &status_, ClientRpcContext::tag(this)); response_reader_->Finish(&response_, &status_, ClientRpcContext::tag(this));
@ -116,8 +119,8 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
} }
ClientRpcContext* StartNewClone() GRPC_OVERRIDE { ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
return new ClientRpcContextUnaryImpl(channel_id_, return new ClientRpcContextUnaryImpl(channel_id_, stub_, req_, start_req_,
stub_, req_, start_req_, callback_); callback_);
} }
private: private:
@ -125,9 +128,9 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
next_state_ = &ClientRpcContextUnaryImpl::DoCallBack; next_state_ = &ClientRpcContextUnaryImpl::DoCallBack;
return false; return false;
} }
bool DoCallBack (bool) { bool DoCallBack(bool) {
callback_(status_, &response_); callback_(status_, &response_);
return true; // we're done, this'll be ignored return true; // we're done, this'll be ignored
} }
grpc::ClientContext context_; grpc::ClientContext context_;
TestService::Stub* stub_; TestService::Stub* stub_;
@ -136,27 +139,28 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
bool (ClientRpcContextUnaryImpl::*next_state_)(bool); bool (ClientRpcContextUnaryImpl::*next_state_)(bool);
std::function<void(grpc::Status, ResponseType*)> callback_; std::function<void(grpc::Status, ResponseType*)> callback_;
std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
TestService::Stub*, grpc::ClientContext*, TestService::Stub*, grpc::ClientContext*, const RequestType&,
const RequestType&, CompletionQueue *)> start_req_; CompletionQueue*)> start_req_;
grpc::Status status_; grpc::Status status_;
double start_; double start_;
std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>> std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>
response_reader_; response_reader_;
}; };
typedef std::forward_list<ClientRpcContext *> context_list; typedef std::forward_list<ClientRpcContext*> context_list;
class AsyncClient : public Client { class AsyncClient : public Client {
public: public:
explicit AsyncClient(const ClientConfig& config, explicit AsyncClient(
std::function<ClientRpcContext*(int, TestService::Stub*, const ClientConfig& config,
const SimpleRequest&)> setup_ctx) : std::function<ClientRpcContext*(int, TestService::Stub*,
Client(config), channel_lock_(config.client_channels()), const SimpleRequest&)> setup_ctx)
contexts_(config.client_channels()), : Client(config),
max_outstanding_per_channel_(config.outstanding_rpcs_per_channel()), channel_lock_(config.client_channels()),
channel_count_(config.client_channels()), contexts_(config.client_channels()),
pref_channel_inc_(config.async_client_threads()) { max_outstanding_per_channel_(config.outstanding_rpcs_per_channel()),
channel_count_(config.client_channels()),
pref_channel_inc_(config.async_client_threads()) {
SetupLoadTest(config, config.async_client_threads()); SetupLoadTest(config, config.async_client_threads());
for (int i = 0; i < config.async_client_threads(); i++) { for (int i = 0; i < config.async_client_threads(); i++) {
@ -181,8 +185,7 @@ class AsyncClient : public Client {
auto ctx = setup_ctx(ch, channel.get_stub(), request_); auto ctx = setup_ctx(ch, channel.get_stub(), request_);
if (closed_loop_) { if (closed_loop_) {
ctx->Start(cq); ctx->Start(cq);
} } else {
else {
contexts_[ch].push_front(ctx); contexts_[ch].push_front(ctx);
} }
} }
@ -210,24 +213,24 @@ class AsyncClient : public Client {
} else { } else {
if (rpc_deadlines_[thread_idx].empty()) { if (rpc_deadlines_[thread_idx].empty()) {
deadline = grpc_time_source::now() + std::chrono::seconds(1); deadline = grpc_time_source::now() + std::chrono::seconds(1);
} } else {
else {
deadline = *(rpc_deadlines_[thread_idx].begin()); deadline = *(rpc_deadlines_[thread_idx].begin());
} }
short_deadline = issue_allowed_[thread_idx] ? short_deadline =
next_issue_[thread_idx] : deadline; issue_allowed_[thread_idx] ? next_issue_[thread_idx] : deadline;
} }
bool got_event; bool got_event;
switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, short_deadline)) { switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, short_deadline)) {
case CompletionQueue::SHUTDOWN: return false; case CompletionQueue::SHUTDOWN:
return false;
case CompletionQueue::TIMEOUT: case CompletionQueue::TIMEOUT:
got_event = false; got_event = false;
break; break;
case CompletionQueue::GOT_EVENT: case CompletionQueue::GOT_EVENT:
got_event = true; got_event = true;
break; break;
default: default:
GPR_ASSERT(false); GPR_ASSERT(false);
break; break;
@ -239,89 +242,87 @@ class AsyncClient : public Client {
return false; return false;
} }
if (got_event) { if (got_event) {
ClientRpcContext* ctx = ClientRpcContext::detag(got_tag); ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
if (ctx->RunNextState(ok, histogram) == false) { if (ctx->RunNextState(ok, histogram) == false) {
// call the callback and then clone the ctx // call the callback and then clone the ctx
ctx->RunNextState(ok, histogram); ctx->RunNextState(ok, histogram);
ClientRpcContext *clone_ctx = ctx->StartNewClone(); ClientRpcContext* clone_ctx = ctx->StartNewClone();
if (closed_loop_) { if (closed_loop_) {
clone_ctx->Start(cli_cqs_[thread_idx].get()); clone_ctx->Start(cli_cqs_[thread_idx].get());
} } else {
else { // Remove the entry from the rpc deadlines list
// Remove the entry from the rpc deadlines list rpc_deadlines_[thread_idx].erase(ctx->deadline_posn());
rpc_deadlines_[thread_idx].erase(ctx->deadline_posn()); // Put the clone_ctx in the list of idle contexts for this channel
// Put the clone_ctx in the list of idle contexts for this channel // Under lock
// Under lock int ch = clone_ctx->channel_id();
int ch = clone_ctx->channel_id(); std::lock_guard<std::mutex> g(channel_lock_[ch]);
std::lock_guard<std::mutex> g(channel_lock_[ch]); contexts_[ch].push_front(clone_ctx);
contexts_[ch].push_front(clone_ctx); }
} // delete the old version
// delete the old version delete ctx;
delete ctx; }
} if (!closed_loop_)
if (!closed_loop_) issue_allowed_[thread_idx] =
issue_allowed_[thread_idx] = true; // may be ok now even if it hadn't been true; // may be ok now even if it hadn't been
} }
if (!closed_loop_ && issue_allowed_[thread_idx] && if (!closed_loop_ && issue_allowed_[thread_idx] &&
grpc_time_source::now() >= next_issue_[thread_idx]) { grpc_time_source::now() >= next_issue_[thread_idx]) {
// Attempt to issue // Attempt to issue
bool issued = false; bool issued = false;
for (int num_attempts = 0, channel_attempt = next_channel_[thread_idx]; for (int num_attempts = 0, channel_attempt = next_channel_[thread_idx];
num_attempts < channel_count_ && !issued; num_attempts++) { num_attempts < channel_count_ && !issued; num_attempts++) {
bool can_issue = false; bool can_issue = false;
ClientRpcContext* ctx = nullptr; ClientRpcContext* ctx = nullptr;
{ {
std::lock_guard<std::mutex> std::lock_guard<std::mutex> g(channel_lock_[channel_attempt]);
g(channel_lock_[channel_attempt]); if (!contexts_[channel_attempt].empty()) {
if (!contexts_[channel_attempt].empty()) { // Get an idle context from the front of the list
// Get an idle context from the front of the list ctx = *(contexts_[channel_attempt].begin());
ctx = *(contexts_[channel_attempt].begin()); contexts_[channel_attempt].pop_front();
contexts_[channel_attempt].pop_front(); can_issue = true;
can_issue = true; }
} }
} if (can_issue) {
if (can_issue) { // do the work to issue
// do the work to issue rpc_deadlines_[thread_idx].emplace_back(grpc_time_source::now() +
rpc_deadlines_[thread_idx].emplace_back( std::chrono::seconds(1));
grpc_time_source::now() + std::chrono::seconds(1)); auto it = rpc_deadlines_[thread_idx].end();
auto it = rpc_deadlines_[thread_idx].end(); --it;
--it; ctx->set_deadline_posn(it);
ctx->set_deadline_posn(it); ctx->Start(cli_cqs_[thread_idx].get());
ctx->Start(cli_cqs_[thread_idx].get()); issued = true;
issued = true; // If we did issue, then next time, try our thread's next
// If we did issue, then next time, try our thread's next // preferred channel
// preferred channel next_channel_[thread_idx] += pref_channel_inc_;
next_channel_[thread_idx] += pref_channel_inc_; if (next_channel_[thread_idx] >= channel_count_)
if (next_channel_[thread_idx] >= channel_count_) next_channel_[thread_idx] = (thread_idx % channel_count_);
next_channel_[thread_idx] = (thread_idx % channel_count_); } else {
} else { // Do a modular increment of channel attempt if we couldn't issue
// Do a modular increment of channel attempt if we couldn't issue channel_attempt = (channel_attempt + 1) % channel_count_;
channel_attempt = (channel_attempt+1) % channel_count_; }
} }
} if (issued) {
if (issued) { // We issued one; see when we can issue the next
// We issued one; see when we can issue the next grpc_time next_issue;
grpc_time next_issue; NextIssueTime(thread_idx, &next_issue);
NextIssueTime(thread_idx, &next_issue); next_issue_[thread_idx] = next_issue;
next_issue_[thread_idx]=next_issue; } else {
} issue_allowed_[thread_idx] = false;
else { }
issue_allowed_[thread_idx] = false; }
} return true;
}
return true;
} }
private: private:
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_; std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
std::vector<deadline_list> rpc_deadlines_; // per thread deadlines std::vector<deadline_list> rpc_deadlines_; // per thread deadlines
std::vector<int> next_channel_; // per thread round-robin channel ctr std::vector<int> next_channel_; // per thread round-robin channel ctr
std::vector<bool> issue_allowed_; // may this thread attempt to issue std::vector<bool> issue_allowed_; // may this thread attempt to issue
std::vector<grpc_time> next_issue_; // when should it issue? std::vector<grpc_time> next_issue_; // when should it issue?
std::vector<std::mutex> channel_lock_; std::vector<std::mutex> channel_lock_;
std::vector<context_list> contexts_; // per-channel list of idle contexts std::vector<context_list> contexts_; // per-channel list of idle contexts
int max_outstanding_per_channel_; int max_outstanding_per_channel_;
int channel_count_; int channel_count_;
int pref_channel_inc_; int pref_channel_inc_;
@ -334,34 +335,31 @@ class AsyncUnaryClient GRPC_FINAL : public AsyncClient {
StartThreads(config.async_client_threads()); StartThreads(config.async_client_threads());
} }
~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); } ~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); }
private:
static ClientRpcContext *SetupCtx(int channel_id, private:
TestService::Stub* stub, static ClientRpcContext* SetupCtx(int channel_id, TestService::Stub* stub,
const SimpleRequest& req) { const SimpleRequest& req) {
auto check_done = [](grpc::Status s, SimpleResponse* response) {}; auto check_done = [](grpc::Status s, SimpleResponse* response) {};
auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx, auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx,
const SimpleRequest& request, CompletionQueue* cq) { const SimpleRequest& request, CompletionQueue* cq) {
return stub->AsyncUnaryCall(ctx, request, cq); return stub->AsyncUnaryCall(ctx, request, cq);
}; };
return new ClientRpcContextUnaryImpl<SimpleRequest, return new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
SimpleResponse>(channel_id, stub, req, channel_id, stub, req, start_req, check_done);
start_req, check_done);
} }
}; };
template <class RequestType, class ResponseType> template <class RequestType, class ResponseType>
class ClientRpcContextStreamingImpl : public ClientRpcContext { class ClientRpcContextStreamingImpl : public ClientRpcContext {
public: public:
ClientRpcContextStreamingImpl(int channel_id, ClientRpcContextStreamingImpl(
TestService::Stub* stub, const RequestType& req, int channel_id, TestService::Stub* stub, const RequestType& req,
std::function<std::unique_ptr< std::function<std::unique_ptr<grpc::ClientAsyncReaderWriter<
grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( RequestType, ResponseType>>(TestService::Stub*, grpc::ClientContext*,
TestService::Stub*, grpc::ClientContext*, CompletionQueue*, CompletionQueue*, void*)> start_req,
void*)> start_req,
std::function<void(grpc::Status, ResponseType*)> on_done) std::function<void(grpc::Status, ResponseType*)> on_done)
: ClientRpcContext(channel_id), : ClientRpcContext(channel_id),
context_(), context_(),
stub_(stub), stub_(stub),
req_(req), req_(req),
response_(), response_(),
@ -374,12 +372,13 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
return (this->*next_state_)(ok, hist); return (this->*next_state_)(ok, hist);
} }
ClientRpcContext* StartNewClone() GRPC_OVERRIDE { ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
return new ClientRpcContextStreamingImpl(channel_id_, return new ClientRpcContextStreamingImpl(channel_id_, stub_, req_,
stub_, req_, start_req_, callback_); start_req_, callback_);
} }
void Start(CompletionQueue *cq) GRPC_OVERRIDE { void Start(CompletionQueue* cq) GRPC_OVERRIDE {
stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this)); stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this));
} }
private: private:
bool ReqSent(bool ok, Histogram*) { return StartWrite(ok); } bool ReqSent(bool ok, Histogram*) { return StartWrite(ok); }
bool StartWrite(bool ok) { bool StartWrite(bool ok) {
@ -411,8 +410,8 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
std::function<void(grpc::Status, ResponseType*)> callback_; std::function<void(grpc::Status, ResponseType*)> callback_;
std::function< std::function<
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
TestService::Stub*, grpc::ClientContext*, TestService::Stub*, grpc::ClientContext*, CompletionQueue*, void*)>
CompletionQueue *, void*)> start_req_; start_req_;
grpc::Status status_; grpc::Status status_;
double start_; double start_;
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>> std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>
@ -430,20 +429,18 @@ class AsyncStreamingClient GRPC_FINAL : public AsyncClient {
} }
~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); } ~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
private:
static ClientRpcContext *SetupCtx(int channel_id, private:
TestService::Stub* stub, static ClientRpcContext* SetupCtx(int channel_id, TestService::Stub* stub,
const SimpleRequest& req) { const SimpleRequest& req) {
auto check_done = [](grpc::Status s, SimpleResponse* response) {}; auto check_done = [](grpc::Status s, SimpleResponse* response) {};
auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx, auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx,
CompletionQueue *cq, void* tag) { CompletionQueue* cq, void* tag) {
auto stream = stub->AsyncStreamingCall(ctx, cq, tag); auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
return stream; return stream;
}; };
return new ClientRpcContextStreamingImpl<SimpleRequest, return new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
SimpleResponse>(channel_id, stub, channel_id, stub, req, start_req, check_done);
req, start_req,
check_done);
} }
}; };

90
test/cpp/qps/interarrival.h
Unescape View File

@ -55,53 +55,60 @@ class RandomDist {
RandomDist() {} RandomDist() {}
virtual ~RandomDist() = 0; virtual ~RandomDist() = 0;
// Argument to operator() is a uniform double in the range [0,1) // Argument to operator() is a uniform double in the range [0,1)
virtual double operator() (double uni) const = 0; virtual double operator()(double uni) const = 0;
}; };
inline RandomDist::~RandomDist() {} inline RandomDist::~RandomDist() {}
class UniformDist GRPC_FINAL: public RandomDist { class UniformDist GRPC_FINAL : public RandomDist {
public: public:
UniformDist(double lo, double hi): lo_(lo), range_(hi-lo) {} UniformDist(double lo, double hi) : lo_(lo), range_(hi - lo) {}
~UniformDist() GRPC_OVERRIDE {} ~UniformDist() GRPC_OVERRIDE {}
double operator() (double uni) const GRPC_OVERRIDE {return uni*range_+lo_;} double operator()(double uni) const GRPC_OVERRIDE {
private: return uni * range_ + lo_;
}
private:
double lo_; double lo_;
double range_; double range_;
}; };
class ExpDist GRPC_FINAL : public RandomDist { class ExpDist GRPC_FINAL : public RandomDist {
public: public:
explicit ExpDist(double lambda): lambda_recip_(1.0/lambda) {} explicit ExpDist(double lambda) : lambda_recip_(1.0 / lambda) {}
~ExpDist() GRPC_OVERRIDE {} ~ExpDist() GRPC_OVERRIDE {}
double operator() (double uni) const GRPC_OVERRIDE { double operator()(double uni) const GRPC_OVERRIDE {
// Note: Use 1.0-uni above to avoid NaN if uni is 0 // Note: Use 1.0-uni above to avoid NaN if uni is 0
return lambda_recip_ * (-log(1.0-uni)); return lambda_recip_ * (-log(1.0 - uni));
} }
private:
private:
double lambda_recip_; double lambda_recip_;
}; };
class DetDist GRPC_FINAL : public RandomDist { class DetDist GRPC_FINAL : public RandomDist {
public: public:
explicit DetDist(double val): val_(val) {} explicit DetDist(double val) : val_(val) {}
~DetDist() GRPC_OVERRIDE {} ~DetDist() GRPC_OVERRIDE {}
double operator() (double uni) const GRPC_OVERRIDE {return val_;} double operator()(double uni) const GRPC_OVERRIDE { return val_; }
private:
private:
double val_; double val_;
}; };
class ParetoDist GRPC_FINAL : public RandomDist { class ParetoDist GRPC_FINAL : public RandomDist {
public: public:
ParetoDist(double base, double alpha): base_(base), alpha_recip_(1.0/alpha) {} ParetoDist(double base, double alpha)
: base_(base), alpha_recip_(1.0 / alpha) {}
~ParetoDist() GRPC_OVERRIDE {} ~ParetoDist() GRPC_OVERRIDE {}
double operator() (double uni) const GRPC_OVERRIDE { double operator()(double uni) const GRPC_OVERRIDE {
// Note: Use 1.0-uni above to avoid div by zero if uni is 0 // Note: Use 1.0-uni above to avoid div by zero if uni is 0
return base_ / pow(1.0-uni, alpha_recip_); return base_ / pow(1.0 - uni, alpha_recip_);
} }
private:
double base_; private:
double alpha_recip_; double base_;
double alpha_recip_;
}; };
// A class library for generating pseudo-random interarrival times // A class library for generating pseudo-random interarrival times
@ -111,38 +118,37 @@ private:
using qps_random_engine = std::default_random_engine; using qps_random_engine = std::default_random_engine;
class InterarrivalTimer { class InterarrivalTimer {
public: public:
InterarrivalTimer() {} InterarrivalTimer() {}
InterarrivalTimer(const RandomDist& r, int threads, int entries=1000000) { InterarrivalTimer(const RandomDist& r, int threads, int entries = 1000000) {
init(r, threads, entries); init(r, threads, entries);
} }
void init(const RandomDist& r, int threads, int entries=1000000) { void init(const RandomDist& r, int threads, int entries = 1000000) {
qps_random_engine gen; qps_random_engine gen;
std::uniform_real_distribution<double> uniform(0.0,1.0); std::uniform_real_distribution<double> uniform(0.0, 1.0);
for (int i=0; i<entries; i++) { for (int i = 0; i < entries; i++) {
random_table_.push_back( random_table_.push_back(std::chrono::nanoseconds(
std::chrono::nanoseconds( static_cast<int64_t>(1e9 * r(uniform(gen)))));
static_cast<int64_t>(1e9*r(uniform(gen)))));
} }
// Now set up the thread positions // Now set up the thread positions
for (int i=0; i<threads; i++) { for (int i = 0; i < threads; i++) {
thread_posns_.push_back(random_table_.begin() + (entries * i)/threads); thread_posns_.push_back(random_table_.begin() + (entries * i) / threads);
} }
} }
virtual ~InterarrivalTimer() {}; virtual ~InterarrivalTimer(){};
std::chrono::nanoseconds operator() (int thread_num) { std::chrono::nanoseconds operator()(int thread_num) {
auto ret = *(thread_posns_[thread_num]++); auto ret = *(thread_posns_[thread_num]++);
if (thread_posns_[thread_num] == random_table_.end()) if (thread_posns_[thread_num] == random_table_.end())
thread_posns_[thread_num] = random_table_.begin(); thread_posns_[thread_num] = random_table_.begin();
return ret; return ret;
} }
private: private:
typedef std::vector<std::chrono::nanoseconds> time_table; typedef std::vector<std::chrono::nanoseconds> time_table;
std::vector<time_table::const_iterator> thread_posns_; std::vector<time_table::const_iterator> thread_posns_;
time_table random_table_; time_table random_table_;
}; };
} }
} }

14
test/cpp/qps/qps_interarrival_test.cc
Unescape View File

@ -42,16 +42,16 @@
using grpc::testing::ExpDist; using grpc::testing::ExpDist;
using grpc::testing::InterarrivalTimer; using grpc::testing::InterarrivalTimer;
void RunTest(InterarrivalTimer&& timer, std::string title) { void RunTest(InterarrivalTimer &&timer, std::string title) {
gpr_histogram *h(gpr_histogram_create(0.01,60e9)); gpr_histogram *h(gpr_histogram_create(0.01, 60e9));
for (int i=0; i<10000000; i++) { for (int i = 0; i < 10000000; i++) {
for (int j=0; j<5; j++) { for (int j = 0; j < 5; j++) {
gpr_histogram_add(h, timer(j).count()); gpr_histogram_add(h, timer(j).count());
} }
} }
std::cout << title << " Distribution" << std::endl; std::cout << title << " Distribution" << std::endl;
std::cout << "Value, Percentile" << std::endl; std::cout << "Value, Percentile" << std::endl;
for (double pct = 0.0; pct < 100.0; pct += 1.0) { for (double pct = 0.0; pct < 100.0; pct += 1.0) {
std::cout << gpr_histogram_percentile(h, pct) << "," << pct << std::endl; std::cout << gpr_histogram_percentile(h, pct) << "," << pct << std::endl;
@ -68,9 +68,9 @@ using grpc::testing::ParetoDist;
int main(int argc, char **argv) { int main(int argc, char **argv) {
RunTest(InterarrivalTimer(ExpDist(10.0), 5), std::string("Exponential(10)")); RunTest(InterarrivalTimer(ExpDist(10.0), 5), std::string("Exponential(10)"));
RunTest(InterarrivalTimer(DetDist(5.0), 5), std::string("Det(5)")); RunTest(InterarrivalTimer(DetDist(5.0), 5), std::string("Det(5)"));
RunTest(InterarrivalTimer(UniformDist(0.0,10.0), 5), RunTest(InterarrivalTimer(UniformDist(0.0, 10.0), 5),
std::string("Uniform(1,10)")); std::string("Uniform(1,10)"));
RunTest(InterarrivalTimer(ParetoDist(1.0,1.0), 5), RunTest(InterarrivalTimer(ParetoDist(1.0, 1.0), 5),
std::string("Pareto(1,1)")); std::string("Pareto(1,1)"));
return 0; return 0;

19
test/cpp/qps/qpstest.proto
Unescape View File

@ -36,7 +36,7 @@ package grpc.testing;
enum PayloadType { enum PayloadType {
// Compressable text format. // Compressable text format.
COMPRESSABLE= 1; COMPRESSABLE = 1;
// Uncompressable binary format. // Uncompressable binary format.
UNCOMPRESSABLE = 2; UNCOMPRESSABLE = 2;
@ -130,20 +130,21 @@ message LoadParams {
message ClientConfig { message ClientConfig {
repeated string server_targets = 1; repeated string server_targets = 1;
required ClientType client_type = 2; required ClientType client_type = 2;
optional bool enable_ssl = 3 [default=false]; optional bool enable_ssl = 3 [default = false];
required int32 outstanding_rpcs_per_channel = 4; required int32 outstanding_rpcs_per_channel = 4;
required int32 client_channels = 5; required int32 client_channels = 5;
required int32 payload_size = 6; required int32 payload_size = 6;
// only for async client: // only for async client:
optional int32 async_client_threads = 7; optional int32 async_client_threads = 7;
optional RpcType rpc_type = 8 [default=UNARY]; optional RpcType rpc_type = 8 [default = UNARY];
optional string host = 9; optional string host = 9;
optional LoadType load_type = 10 [default=CLOSED_LOOP]; optional LoadType load_type = 10 [default = CLOSED_LOOP];
optional LoadParams load_params = 11; optional LoadParams load_params = 11;
} }
// Request current stats // Request current stats
message Mark {} message Mark {
}
message ClientArgs { message ClientArgs {
oneof argtype { oneof argtype {
@ -165,8 +166,8 @@ message ClientStatus {
message ServerConfig { message ServerConfig {
required ServerType server_type = 1; required ServerType server_type = 1;
optional int32 threads = 2 [default=1]; optional int32 threads = 2 [default = 1];
optional bool enable_ssl = 3 [default=false]; optional bool enable_ssl = 3 [default = false];
optional string host = 4; optional string host = 4;
} }
@ -185,11 +186,11 @@ message ServerStatus {
message SimpleRequest { message SimpleRequest {
// Desired payload type in the response from the server. // Desired payload type in the response from the server.
// If response_type is RANDOM, server randomly chooses one from other formats. // If response_type is RANDOM, server randomly chooses one from other formats.
optional PayloadType response_type = 1 [default=COMPRESSABLE]; optional PayloadType response_type = 1 [default = COMPRESSABLE];
// Desired payload size in the response from the server. // Desired payload size in the response from the server.
// If response_type is COMPRESSABLE, this denotes the size before compression. // If response_type is COMPRESSABLE, this denotes the size before compression.
optional int32 response_size = 2 [default=0]; optional int32 response_size = 2 [default = 0];
// Optional input payload sent along with the request. // Optional input payload sent along with the request.
optional Payload payload = 3; optional Payload payload = 3;

Write Preview
Loading…
Cancel
Save