The C based gRPC (C++, Python, Ruby, Objective-C, PHP, C#) https://grpc.io/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

544 lines
18 KiB

10 years ago
/*
*
* Copyright 2015 gRPC authors.
10 years ago
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
10 years ago
*
* http://www.apache.org/licenses/LICENSE-2.0
10 years ago
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
10 years ago
*
*/
#ifndef TEST_QPS_CLIENT_H
#define TEST_QPS_CLIENT_H
#include <stdlib.h>
9 years ago
#include <condition_variable>
#include <mutex>
#include <unordered_map>
#include <vector>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpcpp/channel.h>
#include <grpcpp/support/byte_buffer.h>
#include <grpcpp/support/channel_arguments.h>
#include <grpcpp/support/slice.h>
9 years ago
#include "src/proto/grpc/testing/benchmark_service.grpc.pb.h"
#include "src/proto/grpc/testing/payloads.pb.h"
#include "src/core/lib/gpr/env.h"
#include "src/cpp/util/core_stats.h"
#include "test/cpp/qps/histogram.h"
#include "test/cpp/qps/interarrival.h"
#include "test/cpp/qps/qps_worker.h"
#include "test/cpp/qps/server.h"
#include "test/cpp/qps/usage_timer.h"
9 years ago
#include "test/cpp/util/create_test_channel.h"
#include "test/cpp/util/test_credentials_provider.h"
10 years ago
#define INPROC_NAME_PREFIX "qpsinproc:"
10 years ago
namespace grpc {
namespace testing {
9 years ago
template <class RequestType>
class ClientRequestCreator {
10 years ago
public:
ClientRequestCreator(RequestType* req, const PayloadConfig&) {
// this template must be specialized
// fail with an assertion rather than a compile-time
// check since these only happen at the beginning anyway
GPR_ASSERT(false);
}
};
9 years ago
template <>
class ClientRequestCreator<SimpleRequest> {
public:
ClientRequestCreator(SimpleRequest* req,
const PayloadConfig& payload_config) {
if (payload_config.has_bytebuf_params()) {
GPR_ASSERT(false); // not appropriate for this specialization
} else if (payload_config.has_simple_params()) {
req->set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
req->set_response_size(payload_config.simple_params().resp_size());
req->mutable_payload()->set_type(
9 years ago
grpc::testing::PayloadType::COMPRESSABLE);
int size = payload_config.simple_params().req_size();
std::unique_ptr<char[]> body(new char[size]);
req->mutable_payload()->set_body(body.get(), size);
} else if (payload_config.has_complex_params()) {
GPR_ASSERT(false); // not appropriate for this specialization
} else {
// default should be simple proto without payloads
req->set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
req->set_response_size(0);
req->mutable_payload()->set_type(
grpc::testing::PayloadType::COMPRESSABLE);
}
}
};
9 years ago
template <>
class ClientRequestCreator<ByteBuffer> {
public:
ClientRequestCreator(ByteBuffer* req, const PayloadConfig& payload_config) {
if (payload_config.has_bytebuf_params()) {
std::unique_ptr<char[]> buf(
new char[payload_config.bytebuf_params().req_size()]);
Slice slice(buf.get(), payload_config.bytebuf_params().req_size());
*req = ByteBuffer(&slice, 1);
} else {
GPR_ASSERT(false); // not appropriate for this specialization
9 years ago
}
}
};
9 years ago
class HistogramEntry final {
public:
HistogramEntry() : value_used_(false), status_used_(false) {}
bool value_used() const { return value_used_; }
9 years ago
double value() const { return value_; }
void set_value(double v) {
value_used_ = true;
9 years ago
value_ = v;
}
bool status_used() const { return status_used_; }
int status() const { return status_; }
void set_status(int status) {
status_used_ = true;
status_ = status;
}
9 years ago
private:
bool value_used_;
double value_;
bool status_used_;
int status_;
};
typedef std::unordered_map<int, int64_t> StatusHistogram;
inline void MergeStatusHistogram(const StatusHistogram& from,
StatusHistogram* to) {
for (StatusHistogram::const_iterator it = from.begin(); it != from.end();
++it) {
(*to)[it->first] += it->second;
}
}
10 years ago
class Client {
public:
Client()
: timer_(new UsageTimer),
interarrival_timer_(),
started_requests_(false),
last_reset_poll_count_(0) {
gpr_event_init(&start_requests_);
}
10 years ago
virtual ~Client() {}
ClientStats Mark(bool reset) {
Histogram latencies;
StatusHistogram statuses;
UsageTimer::Result timer_result;
MaybeStartRequests();
int cur_poll_count = GetPollCount();
int poll_count = cur_poll_count - last_reset_poll_count_;
if (reset) {
std::vector<Histogram> to_merge(threads_.size());
std::vector<StatusHistogram> to_merge_status(threads_.size());
for (size_t i = 0; i < threads_.size(); i++) {
threads_[i]->BeginSwap(&to_merge[i], &to_merge_status[i]);
}
std::unique_ptr<UsageTimer> timer(new UsageTimer);
timer_.swap(timer);
for (size_t i = 0; i < threads_.size(); i++) {
latencies.Merge(to_merge[i]);
MergeStatusHistogram(to_merge_status[i], &statuses);
}
timer_result = timer->Mark();
last_reset_poll_count_ = cur_poll_count;
} else {
// merge snapshots of each thread histogram
for (size_t i = 0; i < threads_.size(); i++) {
threads_[i]->MergeStatsInto(&latencies, &statuses);
}
timer_result = timer_->Mark();
}
// Print the median latency per interval for one thread.
// If the number of warmup seconds is x, then the first x + 1 numbers in the
// vector are from the warmup period and should be discarded.
if (median_latency_collection_interval_seconds_ > 0) {
std::vector<double> medians_per_interval =
threads_[0]->GetMedianPerIntervalList();
gpr_log(GPR_INFO, "Num threads: %ld", threads_.size());
gpr_log(GPR_INFO, "Number of medians: %ld", medians_per_interval.size());
for (size_t j = 0; j < medians_per_interval.size(); j++) {
gpr_log(GPR_INFO, "%f", medians_per_interval[j]);
}
}
grpc_stats_data core_stats;
grpc_stats_collect(&core_stats);
ClientStats stats;
latencies.FillProto(stats.mutable_latencies());
for (StatusHistogram::const_iterator it = statuses.begin();
it != statuses.end(); ++it) {
RequestResultCount* rrc = stats.add_request_results();
rrc->set_status_code(it->first);
rrc->set_count(it->second);
}
stats.set_time_elapsed(timer_result.wall);
stats.set_time_system(timer_result.system);
stats.set_time_user(timer_result.user);
stats.set_cq_poll_count(poll_count);
CoreStatsToProto(core_stats, stats.mutable_core_stats());
return stats;
}
// Must call AwaitThreadsCompletion before destructor to avoid a race
// between destructor and invocation of virtual ThreadFunc
void AwaitThreadsCompletion() {
gpr_atm_rel_store(&thread_pool_done_, static_cast<gpr_atm>(true));
DestroyMultithreading();
std::unique_lock<std::mutex> g(thread_completion_mu_);
while (threads_remaining_ != 0) {
threads_complete_.wait(g);
}
}
9 years ago
// Returns the interval (in seconds) between collecting latency medians. If 0,
// no periodic median latencies will be collected.
double GetLatencyCollectionIntervalInSeconds() {
return median_latency_collection_interval_seconds_;
}
virtual int GetPollCount() {
// For sync 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;
10 years ago
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;
10 years ago
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()) {
9 years ago
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)));
}
}
}
gpr_timespec NextIssueTime(int thread_idx) {
const gpr_timespec result = next_time_[thread_idx];
next_time_[thread_idx] =
gpr_time_add(next_time_[thread_idx],
gpr_time_from_nanos(interarrival_timer_.next(thread_idx),
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);
}
class Thread {
public:
Thread(Client* client, size_t idx)
8 years ago
: client_(client), idx_(idx), impl_(&Thread::ThreadFunc, this) {}
8 years ago
~Thread() { impl_.join(); }
void BeginSwap(Histogram* n, StatusHistogram* s) {
std::lock_guard<std::mutex> g(mu_);
n->Swap(&histogram_);
s->swap(statuses_);
}
void MergeStatsInto(Histogram* hist, StatusHistogram* s) {
std::unique_lock<std::mutex> g(mu_);
hist->Merge(histogram_);
MergeStatusHistogram(statuses_, s);
}
std::vector<double> GetMedianPerIntervalList() {
return medians_each_interval_list_;
}
void UpdateHistogram(HistogramEntry* entry) {
std::lock_guard<std::mutex> g(mu_);
if (entry->value_used()) {
histogram_.Add(entry->value());
if (client_->GetLatencyCollectionIntervalInSeconds() > 0) {
histogram_per_interval_.Add(entry->value());
double now = UsageTimer::Now();
if ((now - interval_start_time_) >=
client_->GetLatencyCollectionIntervalInSeconds()) {
// Record the median latency of requests from the last interval.
// Divide by 1e3 to get microseconds.
medians_each_interval_list_.push_back(
histogram_per_interval_.Percentile(50) / 1e3);
histogram_per_interval_.Reset();
interval_start_time_ = now;
}
}
}
if (entry->status_used()) {
statuses_[entry->status()]++;
}
}
private:
Thread(const Thread&);
Thread& operator=(const Thread&);
void ThreadFunc() {
7 years ago
int wait_loop = 0;
while (!gpr_event_wait(
&client_->start_requests_,
gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
7 years ago
gpr_time_from_seconds(20, GPR_TIMESPAN)))) {
gpr_log(GPR_INFO, "%" PRIdPTR ": Waiting for benchmark to start (%d)",
idx_, wait_loop);
wait_loop++;
}
client_->ThreadFunc(idx_, this);
client_->CompleteThread();
}
std::mutex mu_;
Histogram histogram_;
StatusHistogram statuses_;
10 years ago
Client* client_;
const size_t idx_;
std::thread impl_;
// The following are used only if
// median_latency_collection_interval_seconds_ is greater than 0
Histogram histogram_per_interval_;
std::vector<double> medians_each_interval_list_;
double interval_start_time_;
};
bool ThreadCompleted() {
return static_cast<bool>(gpr_atm_acq_load(&thread_pool_done_));
}
virtual void ThreadFunc(size_t thread_idx, Client::Thread* t) = 0;
std::vector<std::unique_ptr<Thread>> threads_;
std::unique_ptr<UsageTimer> timer_;
InterarrivalTimer interarrival_timer_;
std::vector<gpr_timespec> next_time_;
std::mutex thread_completion_mu_;
size_t threads_remaining_;
std::condition_variable threads_complete_;
gpr_event start_requests_;
bool started_requests_;
int last_reset_poll_count_;
void MaybeStartRequests() {
if (!started_requests_) {
started_requests_ = true;
gpr_event_set(&start_requests_, (void*)1);
}
}
void CompleteThread() {
std::lock_guard<std::mutex> g(thread_completion_mu_);
threads_remaining_--;
if (threads_remaining_ == 0) {
threads_complete_.notify_all();
}
}
10 years ago
};
9 years ago
template <class StubType, class RequestType>
class ClientImpl : public Client {
public:
ClientImpl(const ClientConfig& config,
std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)>
create_stub)
: cores_(gpr_cpu_num_cores()), create_stub_(create_stub) {
9 years ago
for (int i = 0; i < config.client_channels(); i++) {
channels_.emplace_back(
config.server_targets(i % config.server_targets_size()), config,
create_stub_, i);
9 years ago
}
7 years ago
std::vector<std::unique_ptr<std::thread>> connecting_threads;
for (auto& c : channels_) {
connecting_threads.emplace_back(c.WaitForReady());
}
for (auto& t : connecting_threads) {
t->join();
}
median_latency_collection_interval_seconds_ =
config.median_latency_collection_interval_millis() / 1e3;
ClientRequestCreator<RequestType> create_req(&request_,
config.payload_config());
9 years ago
}
virtual ~ClientImpl() {}
protected:
const int cores_;
9 years ago
RequestType request_;
class ClientChannelInfo {
public:
ClientChannelInfo(
const grpc::string& target, const ClientConfig& config,
std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)>
create_stub,
int shard) {
ChannelArguments args;
args.SetInt("shard_to_ensure_no_subchannel_merges", shard);
set_channel_args(config, &args);
grpc::string type;
if (config.has_security_params() &&
config.security_params().cred_type().empty()) {
type = kTlsCredentialsType;
} else {
type = config.security_params().cred_type();
}
grpc::string inproc_pfx(INPROC_NAME_PREFIX);
if (target.find(inproc_pfx) != 0) {
channel_ = CreateTestChannel(
target, type, config.security_params().server_host_override(),
!config.security_params().use_test_ca(),
std::shared_ptr<CallCredentials>(), args);
gpr_log(GPR_INFO, "Connecting to %s", target.c_str());
is_inproc_ = false;
} else {
grpc::string tgt = target;
tgt.erase(0, inproc_pfx.length());
int srv_num = std::stoi(tgt);
channel_ = (*g_inproc_servers)[srv_num]->InProcessChannel(args);
is_inproc_ = true;
}
stub_ = create_stub(channel_);
9 years ago
}
Channel* get_channel() { return channel_.get(); }
StubType* get_stub() { return stub_.get(); }
7 years ago
std::unique_ptr<std::thread> WaitForReady() {
return std::unique_ptr<std::thread>(new std::thread([this]() {
if (!is_inproc_) {
int connect_deadline = 10;
/* Allow optionally overriding connect_deadline in order
* to deal with benchmark environments in which the server
* can take a long time to become ready. */
char* channel_connect_timeout_str =
gpr_getenv("QPS_WORKER_CHANNEL_CONNECT_TIMEOUT");
if (channel_connect_timeout_str != nullptr &&
strcmp(channel_connect_timeout_str, "") != 0) {
connect_deadline = atoi(channel_connect_timeout_str);
}
gpr_log(GPR_INFO,
"Waiting for up to %d seconds for the channel %p to connect",
connect_deadline, channel_.get());
gpr_free(channel_connect_timeout_str);
GPR_ASSERT(channel_->WaitForConnected(gpr_time_add(
gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_seconds(connect_deadline, GPR_TIMESPAN))));
gpr_log(GPR_INFO, "Channel %p connected!", channel_.get());
}
7 years ago
}));
}
9 years ago
private:
void set_channel_args(const ClientConfig& config, ChannelArguments* args) {
for (const auto& channel_arg : config.channel_args()) {
if (channel_arg.value_case() == ChannelArg::kStrValue) {
args->SetString(channel_arg.name(), channel_arg.str_value());
} else if (channel_arg.value_case() == ChannelArg::kIntValue) {
args->SetInt(channel_arg.name(), channel_arg.int_value());
} else {
gpr_log(GPR_ERROR, "Empty channel arg value.");
}
}
}
9 years ago
std::shared_ptr<Channel> channel_;
std::unique_ptr<StubType> stub_;
bool is_inproc_;
9 years ago
};
std::vector<ClientChannelInfo> channels_;
std::function<std::unique_ptr<StubType>(const std::shared_ptr<Channel>&)>
create_stub_;
9 years ago
};
std::unique_ptr<Client> CreateSynchronousClient(const ClientConfig& args);
std::unique_ptr<Client> CreateAsyncClient(const ClientConfig& args);
std::unique_ptr<Client> CreateCallbackClient(const ClientConfig& args);
9 years ago
std::unique_ptr<Client> CreateGenericAsyncStreamingClient(
const ClientConfig& args);
10 years ago
} // namespace testing
} // namespace grpc
#endif