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

Merge pull request #17461 from mhaidrygoog/callback_client_streaming_benchmark

Browse Source 
Add support for Callback Client Streaming benchmarks
pull/17634/head
Moiz Haidry 6 years ago committed by GitHub
parent 2dda0bb21b 7bb853ebdd
commit d3db9fee9c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 235 additions and 84 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 118
      test/cpp/qps/client.h
  2. 201
      test/cpp/qps/client_callback.cc

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

@ -236,58 +236,7 @@ class Client {
return 0;
}
protected:
bool closed_loop_;
gpr_atm thread_pool_done_;
double median_latency_collection_interval_seconds_; // In seconds
void StartThreads(size_t num_threads) {
gpr_atm_rel_store(&thread_pool_done_, static_cast<gpr_atm>(false));
threads_remaining_ = num_threads;
for (size_t i = 0; i < num_threads; i++) {
threads_.emplace_back(new Thread(this, i));
}
}
void EndThreads() {
MaybeStartRequests();
threads_.clear();
}
virtual void DestroyMultithreading() = 0;
void SetupLoadTest(const ClientConfig& config, size_t num_threads) {
// Set up the load distribution based on the number of threads
const auto& load = config.load_params();
std::unique_ptr<RandomDistInterface> random_dist;
switch (load.load_case()) {
case LoadParams::kClosedLoop:
// Closed-loop doesn't use random dist at all
break;
case LoadParams::kPoisson:
random_dist.reset(
new ExpDist(load.poisson().offered_load() / num_threads));
break;
default:
GPR_ASSERT(false);
}
// Set closed_loop_ based on whether or not random_dist is set
if (!random_dist) {
closed_loop_ = true;
} else {
closed_loop_ = false;
// set up interarrival timer according to random dist
interarrival_timer_.init(*random_dist, num_threads);
const auto now = gpr_now(GPR_CLOCK_MONOTONIC);
for (size_t i = 0; i < num_threads; i++) {
next_time_.push_back(gpr_time_add(
now,
gpr_time_from_nanos(interarrival_timer_.next(i), GPR_TIMESPAN)));
}
}
}
bool IsClosedLoop() { return closed_loop_; }
gpr_timespec NextIssueTime(int thread_idx) {
const gpr_timespec result = next_time_[thread_idx];
@ -297,9 +246,9 @@ class Client {
GPR_TIMESPAN));
return result;
}
std::function<gpr_timespec()> NextIssuer(int thread_idx) {
return closed_loop_ ? std::function<gpr_timespec()>()
: std::bind(&Client::NextIssueTime, this, thread_idx);
bool ThreadCompleted() {
return static_cast<bool>(gpr_atm_acq_load(&thread_pool_done_));
}
class Thread {
@ -380,8 +329,62 @@ class Client {
double interval_start_time_;
};
bool ThreadCompleted() {
return static_cast<bool>(gpr_atm_acq_load(&thread_pool_done_));
protected:
bool closed_loop_;
gpr_atm thread_pool_done_;
double median_latency_collection_interval_seconds_; // In seconds
void StartThreads(size_t num_threads) {
gpr_atm_rel_store(&thread_pool_done_, static_cast<gpr_atm>(false));
threads_remaining_ = num_threads;
for (size_t i = 0; i < num_threads; i++) {
threads_.emplace_back(new Thread(this, i));
}
}
void EndThreads() {
MaybeStartRequests();
threads_.clear();
}
virtual void DestroyMultithreading() = 0;
void SetupLoadTest(const ClientConfig& config, size_t num_threads) {
// Set up the load distribution based on the number of threads
const auto& load = config.load_params();
std::unique_ptr<RandomDistInterface> random_dist;
switch (load.load_case()) {
case LoadParams::kClosedLoop:
// Closed-loop doesn't use random dist at all
break;
case LoadParams::kPoisson:
random_dist.reset(
new ExpDist(load.poisson().offered_load() / num_threads));
break;
default:
GPR_ASSERT(false);
}
// Set closed_loop_ based on whether or not random_dist is set
if (!random_dist) {
closed_loop_ = true;
} else {
closed_loop_ = false;
// set up interarrival timer according to random dist
interarrival_timer_.init(*random_dist, num_threads);
const auto now = gpr_now(GPR_CLOCK_MONOTONIC);
for (size_t i = 0; i < num_threads; i++) {
next_time_.push_back(gpr_time_add(
now,
gpr_time_from_nanos(interarrival_timer_.next(i), GPR_TIMESPAN)));
}
}
}
std::function<gpr_timespec()> NextIssuer(int thread_idx) {
return closed_loop_ ? std::function<gpr_timespec()>()
: std::bind(&Client::NextIssueTime, this, thread_idx);
}
virtual void ThreadFunc(size_t thread_idx, Client::Thread* t) = 0;
@ -436,6 +439,7 @@ class ClientImpl : public Client {
config.payload_config());
}
virtual ~ClientImpl() {}
const RequestType* request() { return &request_; }
void WaitForChannelsToConnect() {
int connect_deadline_seconds = 10;

201
test/cpp/qps/client_callback.cc
Unescape View File

@ -66,13 +66,35 @@ class CallbackClient
config, BenchmarkStubCreator) {
num_threads_ = NumThreads(config);
rpcs_done_ = 0;
SetupLoadTest(config, num_threads_);
// Don't divide the fixed load among threads as the user threads
// only bootstrap the RPCs
SetupLoadTest(config, 1);
total_outstanding_rpcs_ =
config.client_channels() * config.outstanding_rpcs_per_channel();
}
virtual ~CallbackClient() {}
/**
* The main thread of the benchmark will be waiting on DestroyMultithreading.
* Increment the rpcs_done_ variable to signify that the Callback RPC
* after thread completion is done. When the last outstanding rpc increments
* the counter it should also signal the main thread's conditional variable.
*/
void NotifyMainThreadOfThreadCompletion() {
std::lock_guard<std::mutex> l(shutdown_mu_);
rpcs_done_++;
if (rpcs_done_ == total_outstanding_rpcs_) {
shutdown_cv_.notify_one();
}
}
gpr_timespec NextRPCIssueTime() {
std::lock_guard<std::mutex> l(next_issue_time_mu_);
return Client::NextIssueTime(0);
}
protected:
size_t num_threads_;
size_t total_outstanding_rpcs_;
@ -93,24 +115,9 @@ class CallbackClient
ThreadFuncImpl(t, thread_idx);
}
virtual void ScheduleRpc(Thread* t, size_t thread_idx,
size_t ctx_vector_idx) = 0;
/**
* The main thread of the benchmark will be waiting on DestroyMultithreading.
* Increment the rpcs_done_ variable to signify that the Callback RPC
* after thread completion is done. When the last outstanding rpc increments
* the counter it should also signal the main thread's conditional variable.
*/
void NotifyMainThreadOfThreadCompletion() {
std::lock_guard<std::mutex> l(shutdown_mu_);
rpcs_done_++;
if (rpcs_done_ == total_outstanding_rpcs_) {
shutdown_cv_.notify_one();
}
}
private:
std::mutex next_issue_time_mu_; // Used by next issue time
int NumThreads(const ClientConfig& config) {
int num_threads = config.async_client_threads();
if (num_threads <= 0) { // Use dynamic sizing
@ -149,7 +156,7 @@ class CallbackUnaryClient final : public CallbackClient {
bool ThreadFuncImpl(Thread* t, size_t thread_idx) override {
for (size_t vector_idx = thread_idx; vector_idx < total_outstanding_rpcs_;
vector_idx += num_threads_) {
ScheduleRpc(t, thread_idx, vector_idx);
ScheduleRpc(t, vector_idx);
}
return true;
}
@ -157,26 +164,26 @@ class CallbackUnaryClient final : public CallbackClient {
void InitThreadFuncImpl(size_t thread_idx) override { return; }
private:
void ScheduleRpc(Thread* t, size_t thread_idx, size_t vector_idx) override {
void ScheduleRpc(Thread* t, size_t vector_idx) {
if (!closed_loop_) {
gpr_timespec next_issue_time = NextIssueTime(thread_idx);
gpr_timespec next_issue_time = NextRPCIssueTime();
// Start an alarm callback to run the internal callback after
// next_issue_time
ctx_[vector_idx]->alarm_.experimental().Set(
next_issue_time, [this, t, thread_idx, vector_idx](bool ok) {
IssueUnaryCallbackRpc(t, thread_idx, vector_idx);
next_issue_time, [this, t, vector_idx](bool ok) {
IssueUnaryCallbackRpc(t, vector_idx);
});
} else {
IssueUnaryCallbackRpc(t, thread_idx, vector_idx);
IssueUnaryCallbackRpc(t, vector_idx);
}
}
void IssueUnaryCallbackRpc(Thread* t, size_t thread_idx, size_t vector_idx) {
void IssueUnaryCallbackRpc(Thread* t, size_t vector_idx) {
GPR_TIMER_SCOPE("CallbackUnaryClient::ThreadFunc", 0);
double start = UsageTimer::Now();
ctx_[vector_idx]->stub_->experimental_async()->UnaryCall(
(&ctx_[vector_idx]->context_), &request_, &ctx_[vector_idx]->response_,
[this, t, thread_idx, start, vector_idx](grpc::Status s) {
[this, t, start, vector_idx](grpc::Status s) {
// Update Histogram with data from the callback run
HistogramEntry entry;
if (s.ok()) {
@ -193,17 +200,157 @@ class CallbackUnaryClient final : public CallbackClient {
ctx_[vector_idx].reset(
new CallbackClientRpcContext(ctx_[vector_idx]->stub_));
// Schedule a new RPC
ScheduleRpc(t, thread_idx, vector_idx);
ScheduleRpc(t, vector_idx);
}
});
}
};
class CallbackStreamingClient : public CallbackClient {
public:
CallbackStreamingClient(const ClientConfig& config)
: CallbackClient(config),
messages_per_stream_(config.messages_per_stream()) {
for (int ch = 0; ch < config.client_channels(); ch++) {
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
ctx_.emplace_back(
new CallbackClientRpcContext(channels_[ch].get_stub()));
}
}
StartThreads(num_threads_);
}
~CallbackStreamingClient() {}
void AddHistogramEntry(double start_, bool ok, Thread* thread_ptr) {
// Update Histogram with data from the callback run
HistogramEntry entry;
if (ok) {
entry.set_value((UsageTimer::Now() - start_) * 1e9);
}
thread_ptr->UpdateHistogram(&entry);
}
int messages_per_stream() { return messages_per_stream_; }
protected:
const int messages_per_stream_;
};
class CallbackStreamingPingPongClient : public CallbackStreamingClient {
public:
CallbackStreamingPingPongClient(const ClientConfig& config)
: CallbackStreamingClient(config) {}
~CallbackStreamingPingPongClient() {}
};
class CallbackStreamingPingPongReactor final
: public grpc::experimental::ClientBidiReactor<SimpleRequest,
SimpleResponse> {
public:
CallbackStreamingPingPongReactor(
CallbackStreamingPingPongClient* client,
std::unique_ptr<CallbackClientRpcContext> ctx)
: client_(client), ctx_(std::move(ctx)), messages_issued_(0) {}
void StartNewRpc() {
if (client_->ThreadCompleted()) return;
start_ = UsageTimer::Now();
ctx_->stub_->experimental_async()->StreamingCall(&(ctx_->context_), this);
StartWrite(client_->request());
StartCall();
}
void OnWriteDone(bool ok) override {
if (!ok || client_->ThreadCompleted()) {
if (!ok) gpr_log(GPR_ERROR, "Error writing RPC");
StartWritesDone();
return;
}
StartRead(&ctx_->response_);
}
void OnReadDone(bool ok) override {
client_->AddHistogramEntry(start_, ok, thread_ptr_);
if (client_->ThreadCompleted() || !ok ||
(client_->messages_per_stream() != 0 &&
++messages_issued_ >= client_->messages_per_stream())) {
if (!ok) {
gpr_log(GPR_ERROR, "Error reading RPC");
}
StartWritesDone();
return;
}
StartWrite(client_->request());
}
void OnDone(const Status& s) override {
if (client_->ThreadCompleted() || !s.ok()) {
client_->NotifyMainThreadOfThreadCompletion();
return;
}
ctx_.reset(new CallbackClientRpcContext(ctx_->stub_));
ScheduleRpc();
}
void ScheduleRpc() {
if (client_->ThreadCompleted()) return;
if (!client_->IsClosedLoop()) {
gpr_timespec next_issue_time = client_->NextRPCIssueTime();
// Start an alarm callback to run the internal callback after
// next_issue_time
ctx_->alarm_.experimental().Set(next_issue_time,
[this](bool ok) { StartNewRpc(); });
} else {
StartNewRpc();
}
}
void set_thread_ptr(Client::Thread* ptr) { thread_ptr_ = ptr; }
CallbackStreamingPingPongClient* client_;
std::unique_ptr<CallbackClientRpcContext> ctx_;
Client::Thread* thread_ptr_; // Needed to update histogram entries
double start_; // Track message start time
int messages_issued_; // Messages issued by this stream
};
class CallbackStreamingPingPongClientImpl final
: public CallbackStreamingPingPongClient {
public:
CallbackStreamingPingPongClientImpl(const ClientConfig& config)
: CallbackStreamingPingPongClient(config) {
for (size_t i = 0; i < total_outstanding_rpcs_; i++)
reactor_.emplace_back(
new CallbackStreamingPingPongReactor(this, std::move(ctx_[i])));
}
~CallbackStreamingPingPongClientImpl() {}
bool ThreadFuncImpl(Client::Thread* t, size_t thread_idx) override {
for (size_t vector_idx = thread_idx; vector_idx < total_outstanding_rpcs_;
vector_idx += num_threads_) {
reactor_[vector_idx]->set_thread_ptr(t);
reactor_[vector_idx]->ScheduleRpc();
}
return true;
}
void InitThreadFuncImpl(size_t thread_idx) override {}
private:
std::vector<std::unique_ptr<CallbackStreamingPingPongReactor>> reactor_;
};
// TODO(mhaidry) : Implement Streaming from client, server and both ways
std::unique_ptr<Client> CreateCallbackClient(const ClientConfig& config) {
switch (config.rpc_type()) {
case UNARY:
return std::unique_ptr<Client>(new CallbackUnaryClient(config));
case STREAMING:
return std::unique_ptr<Client>(
new CallbackStreamingPingPongClientImpl(config));
case STREAMING_FROM_CLIENT:
case STREAMING_FROM_SERVER:
case STREAMING_BOTH_WAYS:

Write Preview
Loading…
Cancel
Save