aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp/qps
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp/qps')
-rw-r--r--test/cpp/qps/client.h11
-rw-r--r--test/cpp/qps/client_async.cc42
-rw-r--r--test/cpp/qps/client_sync.cc24
-rw-r--r--test/cpp/qps/driver.cc200
-rw-r--r--test/cpp/qps/driver.h4
-rwxr-xr-xtest/cpp/qps/gen_build_yaml.py63
-rw-r--r--test/cpp/qps/interarrival.h6
-rw-r--r--test/cpp/qps/json_run_localhost.cc73
-rw-r--r--test/cpp/qps/qps_json_driver.cc150
-rw-r--r--test/cpp/qps/qps_worker.cc20
-rw-r--r--test/cpp/qps/report.cc15
-rw-r--r--test/cpp/qps/report.h30
-rw-r--r--test/cpp/qps/server.h2
-rw-r--r--test/cpp/qps/server_async.cc24
-rw-r--r--test/cpp/qps/server_sync.cc8
-rw-r--r--test/cpp/qps/usage_timer.cc33
-rw-r--r--test/cpp/qps/usage_timer.h2
17 files changed, 479 insertions, 228 deletions
diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h
index 9983c8a7b0..fdd78ebb89 100644
--- a/test/cpp/qps/client.h
+++ b/test/cpp/qps/client.h
@@ -103,7 +103,7 @@ class ClientRequestCreator<ByteBuffer> {
if (payload_config.has_bytebuf_params()) {
std::unique_ptr<char[]> buf(
new char[payload_config.bytebuf_params().req_size()]);
- gpr_slice s = gpr_slice_from_copied_buffer(
+ grpc_slice s = grpc_slice_from_copied_buffer(
buf.get(), payload_config.bytebuf_params().req_size());
Slice slice(s, Slice::STEAL_REF);
*req = ByteBuffer(&slice, 1);
@@ -113,7 +113,7 @@ class ClientRequestCreator<ByteBuffer> {
}
};
-class HistogramEntry GRPC_FINAL {
+class HistogramEntry final {
public:
HistogramEntry() : value_used_(false), status_used_(false) {}
bool value_used() const { return value_used_; }
@@ -163,10 +163,9 @@ class Client {
MaybeStartRequests();
- // avoid std::vector for old compilers that expect a copy constructor
if (reset) {
- Histogram* to_merge = new Histogram[threads_.size()];
- StatusHistogram* to_merge_status = new StatusHistogram[threads_.size()];
+ 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]);
@@ -177,8 +176,6 @@ class Client {
latencies.Merge(to_merge[i]);
MergeStatusHistogram(to_merge_status[i], &statuses);
}
- delete[] to_merge;
- delete[] to_merge_status;
timer_result = timer->Mark();
} else {
// merge snapshots of each thread histogram
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 4d36a6ba42..4032039ea1 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -93,8 +93,8 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
callback_(on_done),
next_issue_(next_issue),
start_req_(start_req) {}
- ~ClientRpcContextUnaryImpl() GRPC_OVERRIDE {}
- void Start(CompletionQueue* cq) GRPC_OVERRIDE {
+ ~ClientRpcContextUnaryImpl() override {}
+ void Start(CompletionQueue* cq) override {
cq_ = cq;
if (!next_issue_) { // ready to issue
RunNextState(true, nullptr);
@@ -102,7 +102,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
alarm_.reset(new Alarm(cq_, next_issue_(), ClientRpcContext::tag(this)));
}
}
- bool RunNextState(bool ok, HistogramEntry* entry) GRPC_OVERRIDE {
+ bool RunNextState(bool ok, HistogramEntry* entry) override {
switch (next_state_) {
case State::READY:
start_ = UsageTimer::Now();
@@ -121,7 +121,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
return false;
}
}
- ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
+ ClientRpcContext* StartNewClone() override {
return new ClientRpcContextUnaryImpl(stub_, req_, next_issue_, start_req_,
callback_);
}
@@ -177,7 +177,6 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
shutdown_state_.emplace_back(new PerThreadShutdownState());
}
- using namespace std::placeholders;
int t = 0;
for (int ch = 0; ch < config.client_channels(); ch++) {
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
@@ -217,7 +216,7 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
}
return num_threads;
}
- void DestroyMultithreading() GRPC_OVERRIDE GRPC_FINAL {
+ void DestroyMultithreading() override final {
for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
std::lock_guard<std::mutex> lock((*ss)->mutex);
(*ss)->shutdown = true;
@@ -228,8 +227,7 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
this->EndThreads(); // this needed for resolution
}
- bool ThreadFunc(HistogramEntry* entry,
- size_t thread_idx) GRPC_OVERRIDE GRPC_FINAL {
+ bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) override final {
void* got_tag;
bool ok;
@@ -279,7 +277,7 @@ static std::unique_ptr<BenchmarkService::Stub> BenchmarkStubCreator(
return BenchmarkService::NewStub(ch);
}
-class AsyncUnaryClient GRPC_FINAL
+class AsyncUnaryClient final
: public AsyncClient<BenchmarkService::Stub, SimpleRequest> {
public:
explicit AsyncUnaryClient(const ClientConfig& config)
@@ -287,7 +285,7 @@ class AsyncUnaryClient GRPC_FINAL
config, SetupCtx, BenchmarkStubCreator) {
StartThreads(num_async_threads_);
}
- ~AsyncUnaryClient() GRPC_OVERRIDE {}
+ ~AsyncUnaryClient() override {}
private:
static void CheckDone(grpc::Status s, SimpleResponse* response,
@@ -329,13 +327,13 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
callback_(on_done),
next_issue_(next_issue),
start_req_(start_req) {}
- ~ClientRpcContextStreamingImpl() GRPC_OVERRIDE {}
- void Start(CompletionQueue* cq) GRPC_OVERRIDE {
+ ~ClientRpcContextStreamingImpl() override {}
+ void Start(CompletionQueue* cq) override {
cq_ = cq;
stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this));
next_state_ = State::STREAM_IDLE;
}
- bool RunNextState(bool ok, HistogramEntry* entry) GRPC_OVERRIDE {
+ bool RunNextState(bool ok, HistogramEntry* entry) override {
while (true) {
switch (next_state_) {
case State::STREAM_IDLE:
@@ -377,7 +375,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
}
}
}
- ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
+ ClientRpcContext* StartNewClone() override {
return new ClientRpcContextStreamingImpl(stub_, req_, next_issue_,
start_req_, callback_);
}
@@ -410,7 +408,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
stream_;
};
-class AsyncStreamingClient GRPC_FINAL
+class AsyncStreamingClient final
: public AsyncClient<BenchmarkService::Stub, SimpleRequest> {
public:
explicit AsyncStreamingClient(const ClientConfig& config)
@@ -419,7 +417,7 @@ class AsyncStreamingClient GRPC_FINAL
StartThreads(num_async_threads_);
}
- ~AsyncStreamingClient() GRPC_OVERRIDE {}
+ ~AsyncStreamingClient() override {}
private:
static void CheckDone(grpc::Status s, SimpleResponse* response) {}
@@ -458,8 +456,8 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
callback_(on_done),
next_issue_(next_issue),
start_req_(start_req) {}
- ~ClientRpcContextGenericStreamingImpl() GRPC_OVERRIDE {}
- void Start(CompletionQueue* cq) GRPC_OVERRIDE {
+ ~ClientRpcContextGenericStreamingImpl() override {}
+ void Start(CompletionQueue* cq) override {
cq_ = cq;
const grpc::string kMethodName(
"/grpc.testing.BenchmarkService/StreamingCall");
@@ -467,7 +465,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
ClientRpcContext::tag(this));
next_state_ = State::STREAM_IDLE;
}
- bool RunNextState(bool ok, HistogramEntry* entry) GRPC_OVERRIDE {
+ bool RunNextState(bool ok, HistogramEntry* entry) override {
while (true) {
switch (next_state_) {
case State::STREAM_IDLE:
@@ -509,7 +507,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
}
}
}
- ClientRpcContext* StartNewClone() GRPC_OVERRIDE {
+ ClientRpcContext* StartNewClone() override {
return new ClientRpcContextGenericStreamingImpl(stub_, req_, next_issue_,
start_req_, callback_);
}
@@ -546,7 +544,7 @@ static std::unique_ptr<grpc::GenericStub> GenericStubCreator(
return std::unique_ptr<grpc::GenericStub>(new grpc::GenericStub(ch));
}
-class GenericAsyncStreamingClient GRPC_FINAL
+class GenericAsyncStreamingClient final
: public AsyncClient<grpc::GenericStub, ByteBuffer> {
public:
explicit GenericAsyncStreamingClient(const ClientConfig& config)
@@ -555,7 +553,7 @@ class GenericAsyncStreamingClient GRPC_FINAL
StartThreads(num_async_threads_);
}
- ~GenericAsyncStreamingClient() GRPC_OVERRIDE {}
+ ~GenericAsyncStreamingClient() override {}
private:
static void CheckDone(grpc::Status s, ByteBuffer* response) {}
diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc
index f61e80d76b..b1e61865e7 100644
--- a/test/cpp/qps/client_sync.cc
+++ b/test/cpp/qps/client_sync.cc
@@ -108,10 +108,10 @@ class SynchronousClient
std::vector<SimpleResponse> responses_;
private:
- void DestroyMultithreading() GRPC_OVERRIDE GRPC_FINAL { EndThreads(); }
+ void DestroyMultithreading() override final { EndThreads(); }
};
-class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
+class SynchronousUnaryClient final : public SynchronousClient {
public:
SynchronousUnaryClient(const ClientConfig& config)
: SynchronousClient(config) {
@@ -119,7 +119,7 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
}
~SynchronousUnaryClient() {}
- bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) GRPC_OVERRIDE {
+ bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) override {
if (!WaitToIssue(thread_idx)) {
return true;
}
@@ -135,13 +135,12 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
}
};
-class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
+class SynchronousStreamingClient final : public SynchronousClient {
public:
SynchronousStreamingClient(const ClientConfig& config)
- : SynchronousClient(config) {
- context_ = new grpc::ClientContext[num_threads_];
- stream_ = new std::unique_ptr<
- grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>[num_threads_];
+ : SynchronousClient(config),
+ context_(num_threads_),
+ stream_(num_threads_) {
for (size_t thread_idx = 0; thread_idx < num_threads_; thread_idx++) {
auto* stub = channels_[thread_idx % channels_.size()].get_stub();
stream_[thread_idx] = stub->StreamingCall(&context_[thread_idx]);
@@ -161,11 +160,9 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
}
}
}
- delete[] stream_;
- delete[] context_;
}
- bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) GRPC_OVERRIDE {
+ bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) override {
if (!WaitToIssue(thread_idx)) {
return true;
}
@@ -182,8 +179,9 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
private:
// These are both conceptually std::vector but cannot be for old compilers
// that expect contained classes to support copy constructors
- grpc::ClientContext* context_;
- std::unique_ptr<grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>*
+ std::vector<grpc::ClientContext> context_;
+ std::vector<
+ std::unique_ptr<grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>>
stream_;
};
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index a440341ccf..93ef32db77 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -44,6 +44,7 @@
#include <grpc/support/alloc.h>
#include <grpc/support/host_port.h>
#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
#include "src/core/lib/profiling/timers.h"
#include "src/core/lib/support/env.h"
@@ -99,23 +100,36 @@ static std::unordered_map<string, std::deque<int>> get_hosts_and_cores(
return hosts;
}
-static deque<string> get_workers(const string& name) {
- char* env = gpr_getenv(name.c_str());
- if (!env) return deque<string>();
-
+static deque<string> get_workers(const string& env_name) {
+ char* env = gpr_getenv(env_name.c_str());
+ if (!env) {
+ env = gpr_strdup("");
+ }
deque<string> out;
char* p = env;
- for (;;) {
- char* comma = strchr(p, ',');
- if (comma) {
- out.emplace_back(p, comma);
- p = comma + 1;
- } else {
- out.emplace_back(p);
- gpr_free(env);
- return out;
+ if (strlen(env) != 0) {
+ for (;;) {
+ char* comma = strchr(p, ',');
+ if (comma) {
+ out.emplace_back(p, comma);
+ p = comma + 1;
+ } else {
+ out.emplace_back(p);
+ break;
+ }
}
}
+ if (out.size() == 0) {
+ gpr_log(GPR_ERROR,
+ "Environment variable \"%s\" does not contain a list of QPS "
+ "workers to use. Set it to a comma-separated list of "
+ "hostname:port pairs, starting with hosts that should act as "
+ "servers. E.g. export "
+ "%s=\"serverhost1:1234,clienthost1:1234,clienthost2:1234\"",
+ env_name.c_str(), env_name.c_str());
+ }
+ gpr_free(env);
+ return out;
}
// helpers for postprocess_scenario_result
@@ -125,6 +139,8 @@ static double UserTime(ClientStats s) { return s.time_user(); }
static double ServerWallTime(ServerStats s) { return s.time_elapsed(); }
static double ServerSystemTime(ServerStats s) { return s.time_system(); }
static double ServerUserTime(ServerStats s) { return s.time_user(); }
+static double ServerTotalCpuTime(ServerStats s) { return s.total_cpu_time(); }
+static double ServerIdleCpuTime(ServerStats s) { return s.idle_cpu_time(); }
static int Cores(int n) { return n; }
// Postprocess ScenarioResult and populate result summary.
@@ -149,6 +165,7 @@ static void postprocess_scenario_result(ScenarioResult* result) {
sum(result->server_stats(), ServerWallTime);
auto server_user_time = 100.0 * sum(result->server_stats(), ServerUserTime) /
sum(result->server_stats(), ServerWallTime);
+
auto client_system_time = 100.0 * sum(result->client_stats(), SystemTime) /
sum(result->client_stats(), WallTime);
auto client_user_time = 100.0 * sum(result->client_stats(), UserTime) /
@@ -159,6 +176,18 @@ static void postprocess_scenario_result(ScenarioResult* result) {
result->mutable_summary()->set_client_system_time(client_system_time);
result->mutable_summary()->set_client_user_time(client_user_time);
+ // For Non-linux platform, get_cpu_usage() is not implemented. Thus,
+ // ServerTotalCpuTime and ServerIdleCpuTime are both 0.
+ if (average(result->server_stats(), ServerTotalCpuTime) == 0) {
+ result->mutable_summary()->set_server_cpu_usage(0);
+ } else {
+ auto server_cpu_usage =
+ 100 -
+ 100 * average(result->server_stats(), ServerIdleCpuTime) /
+ average(result->server_stats(), ServerTotalCpuTime);
+ result->mutable_summary()->set_server_cpu_usage(server_cpu_usage);
+ }
+
if (result->request_results_size() > 0) {
int64_t successes = 0;
int64_t failures = 0;
@@ -177,39 +206,22 @@ static void postprocess_scenario_result(ScenarioResult* result) {
}
}
-// Namespace for classes and functions used only in RunScenario
-// Using this rather than local definitions to workaround gcc-4.4 limitations
-// regarding using templates without linkage
-namespace runsc {
-
-// ClientContext allocator
-static ClientContext* AllocContext(list<ClientContext>* contexts) {
- contexts->emplace_back();
- auto context = &contexts->back();
- context->set_wait_for_ready(true);
- return context;
-}
-
-struct ServerData {
- unique_ptr<WorkerService::Stub> stub;
- unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
-};
-
-struct ClientData {
- unique_ptr<WorkerService::Stub> stub;
- unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
-};
-} // namespace runsc
-
std::unique_ptr<ScenarioResult> RunScenario(
const ClientConfig& initial_client_config, size_t num_clients,
const ServerConfig& initial_server_config, size_t num_servers,
- int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count) {
+ int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
+ const char* qps_server_target_override, bool configure_core_lists) {
// Log everything from the driver
gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
// ClientContext allocations (all are destroyed at scope exit)
list<ClientContext> contexts;
+ auto alloc_context = [](list<ClientContext>* contexts) {
+ contexts->emplace_back();
+ auto context = &contexts->back();
+ context->set_wait_for_ready(true);
+ return context;
+ };
// To be added to the result, containing the final configuration used for
// client and config (including host, etc.)
@@ -243,9 +255,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
workers.push_back(addr);
}
}
-
- // Setup the hosts and core counts
- auto hosts_cores = get_hosts_and_cores(workers);
+ GPR_ASSERT(workers.size() != 0);
// if num_clients is set to <=0, do dynamic sizing: all workers
// except for servers are clients
@@ -262,10 +272,16 @@ std::unique_ptr<ScenarioResult> RunScenario(
workers.resize(num_clients + num_servers);
// Start servers
- using runsc::ServerData;
- // servers is array rather than std::vector to avoid gcc-4.4 issues
- // where class contained in std::vector must have a copy constructor
- auto* servers = new ServerData[num_servers];
+ struct ServerData {
+ unique_ptr<WorkerService::Stub> stub;
+ unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
+ };
+ std::vector<ServerData> servers(num_servers);
+ std::unordered_map<string, std::deque<int>> hosts_cores;
+
+ if (configure_core_lists) {
+ hosts_cores = get_hosts_and_cores(workers);
+ }
for (size_t i = 0; i < num_servers; i++) {
gpr_log(GPR_INFO, "Starting server on %s (worker #%" PRIuPTR ")",
workers[i].c_str(), i);
@@ -273,44 +289,42 @@ std::unique_ptr<ScenarioResult> RunScenario(
CreateChannel(workers[i], InsecureChannelCredentials()));
ServerConfig server_config = initial_server_config;
- char* host;
- char* driver_port;
- char* cli_target;
- gpr_split_host_port(workers[i].c_str(), &host, &driver_port);
- string host_str(host);
int server_core_limit = initial_server_config.core_limit();
int client_core_limit = initial_client_config.core_limit();
- if (server_core_limit == 0 && client_core_limit > 0) {
- // In this case, limit the server cores if it matches the
- // same host as one or more clients
- const auto& dq = hosts_cores.at(host_str);
- bool match = false;
- int limit = dq.size();
- for (size_t cli = 0; cli < num_clients; cli++) {
- if (host_str == get_host(workers[cli + num_servers])) {
- limit -= client_core_limit;
- match = true;
+ if (configure_core_lists) {
+ string host_str(get_host(workers[i]));
+ if (server_core_limit == 0 && client_core_limit > 0) {
+ // In this case, limit the server cores if it matches the
+ // same host as one or more clients
+ const auto& dq = hosts_cores.at(host_str);
+ bool match = false;
+ int limit = dq.size();
+ for (size_t cli = 0; cli < num_clients; cli++) {
+ if (host_str == get_host(workers[cli + num_servers])) {
+ limit -= client_core_limit;
+ match = true;
+ }
+ }
+ if (match) {
+ GPR_ASSERT(limit > 0);
+ server_core_limit = limit;
}
}
- if (match) {
- GPR_ASSERT(limit > 0);
- server_core_limit = limit;
- }
- }
- if (server_core_limit > 0) {
- auto& dq = hosts_cores.at(host_str);
- GPR_ASSERT(dq.size() >= static_cast<size_t>(server_core_limit));
- for (int core = 0; core < server_core_limit; core++) {
- server_config.add_core_list(dq.front());
- dq.pop_front();
+ if (server_core_limit > 0) {
+ auto& dq = hosts_cores.at(host_str);
+ GPR_ASSERT(dq.size() >= static_cast<size_t>(server_core_limit));
+ gpr_log(GPR_INFO, "Setting server core_list");
+ for (int core = 0; core < server_core_limit; core++) {
+ server_config.add_core_list(dq.front());
+ dq.pop_front();
+ }
}
}
ServerArgs args;
*args.mutable_setup() = server_config;
- servers[i].stream =
- servers[i].stub->RunServer(runsc::AllocContext(&contexts));
+ servers[i].stream = servers[i].stub->RunServer(alloc_context(&contexts));
if (!servers[i].stream->Write(args)) {
gpr_log(GPR_ERROR, "Could not write args to server %zu", i);
}
@@ -318,20 +332,29 @@ std::unique_ptr<ScenarioResult> RunScenario(
if (!servers[i].stream->Read(&init_status)) {
gpr_log(GPR_ERROR, "Server %zu did not yield initial status", i);
}
- gpr_join_host_port(&cli_target, host, init_status.port());
- client_config.add_server_targets(cli_target);
- gpr_free(host);
- gpr_free(driver_port);
- gpr_free(cli_target);
+ if (qps_server_target_override != NULL &&
+ strlen(qps_server_target_override) > 0) {
+ // overriding the qps server target only works if there is 1 server
+ GPR_ASSERT(num_servers == 1);
+ client_config.add_server_targets(qps_server_target_override);
+ } else {
+ std::string host;
+ char* cli_target;
+ host = get_host(workers[i]);
+ gpr_join_host_port(&cli_target, host.c_str(), init_status.port());
+ client_config.add_server_targets(cli_target);
+ gpr_free(cli_target);
+ }
}
// Targets are all set by now
result_client_config = client_config;
// Start clients
- using runsc::ClientData;
- // clients is array rather than std::vector to avoid gcc-4.4 issues
- // where class contained in std::vector must have a copy constructor
- auto* clients = new ClientData[num_clients];
+ struct ClientData {
+ unique_ptr<WorkerService::Stub> stub;
+ unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
+ };
+ std::vector<ClientData> clients(num_clients);
size_t channels_allocated = 0;
for (size_t i = 0; i < num_clients; i++) {
const auto& worker = workers[i + num_servers];
@@ -343,7 +366,8 @@ std::unique_ptr<ScenarioResult> RunScenario(
int server_core_limit = initial_server_config.core_limit();
int client_core_limit = initial_client_config.core_limit();
- if ((server_core_limit > 0) || (client_core_limit > 0)) {
+ if (configure_core_lists &&
+ ((server_core_limit > 0) || (client_core_limit > 0))) {
auto& dq = hosts_cores.at(get_host(worker));
if (client_core_limit == 0) {
// limit client cores if it matches a server host
@@ -361,6 +385,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
}
if (client_core_limit > 0) {
GPR_ASSERT(dq.size() >= static_cast<size_t>(client_core_limit));
+ gpr_log(GPR_INFO, "Setting client core_list");
for (int core = 0; core < client_core_limit; core++) {
per_client_config.add_core_list(dq.front());
dq.pop_front();
@@ -380,8 +405,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
ClientArgs args;
*args.mutable_setup() = per_client_config;
- clients[i].stream =
- clients[i].stub->RunClient(runsc::AllocContext(&contexts));
+ clients[i].stream = clients[i].stub->RunClient(alloc_context(&contexts));
if (!clients[i].stream->Write(args)) {
gpr_log(GPR_ERROR, "Could not write args to client %zu", i);
}
@@ -501,7 +525,6 @@ std::unique_ptr<ScenarioResult> RunScenario(
s.error_message().c_str());
}
}
- delete[] clients;
merged_latencies.FillProto(result->mutable_latencies());
for (std::unordered_map<int, int64_t>::iterator it = merged_statuses.begin();
@@ -544,8 +567,6 @@ std::unique_ptr<ScenarioResult> RunScenario(
}
}
- delete[] servers;
-
postprocess_scenario_result(result.get());
return result;
}
@@ -554,6 +575,9 @@ bool RunQuit() {
// Get client, server lists
bool result = true;
auto workers = get_workers("QPS_WORKERS");
+ if (workers.size() == 0) {
+ return false;
+ }
for (size_t i = 0; i < workers.size(); i++) {
auto stub = WorkerService::NewStub(
CreateChannel(workers[i], InsecureChannelCredentials()));
diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h
index 93f4370caf..b5c8152e1b 100644
--- a/test/cpp/qps/driver.h
+++ b/test/cpp/qps/driver.h
@@ -45,7 +45,9 @@ namespace testing {
std::unique_ptr<ScenarioResult> RunScenario(
const grpc::testing::ClientConfig& client_config, size_t num_clients,
const grpc::testing::ServerConfig& server_config, size_t num_servers,
- int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count);
+ int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
+ const char* qps_server_target_override = "",
+ bool configure_core_lists = true);
bool RunQuit();
} // namespace testing
diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py
index 369da2c8ca..188d6196e5 100755
--- a/test/cpp/qps/gen_build_yaml.py
+++ b/test/cpp/qps/gen_build_yaml.py
@@ -43,28 +43,38 @@ sys.path.append(run_tests_root)
import performance.scenario_config as scenario_config
-def _scenario_json_string(scenario_json):
+configs_from_yaml = yaml.load(open(os.path.join(os.path.dirname(sys.argv[0]), '../../../build.yaml')))['configs'].keys()
+
+def mutate_scenario(scenario_json, is_tsan):
# tweak parameters to get fast test times
+ scenario_json = dict(scenario_json)
scenario_json['warmup_seconds'] = 0
scenario_json['benchmark_seconds'] = 1
- scenarios_json = {'scenarios': [scenario_config.remove_nonproto_fields(scenario_json)]}
+ outstanding_rpcs_divisor = 1
+ if is_tsan and (
+ scenario_json['client_config']['client_type'] == 'SYNC_CLIENT' or
+ scenario_json['server_config']['server_type'] == 'SYNC_SERVER'):
+ outstanding_rpcs_divisor = 10
+ scenario_json['client_config']['outstanding_rpcs_per_channel'] = max(1,
+ int(scenario_json['client_config']['outstanding_rpcs_per_channel'] / outstanding_rpcs_divisor))
+ return scenario_json
+
+def _scenario_json_string(scenario_json, is_tsan):
+ scenarios_json = {'scenarios': [scenario_config.remove_nonproto_fields(mutate_scenario(scenario_json, is_tsan))]}
return json.dumps(scenarios_json)
-def threads_of_type(scenario_json, path):
- d = scenario_json
- for el in path.split('/'):
- if el not in d:
- return 0
- d = d[el]
- return d
+def threads_required(scenario_json, where, is_tsan):
+ scenario_json = mutate_scenario(scenario_json, is_tsan)
+ if scenario_json['%s_config' % where]['%s_type' % where] == 'ASYNC_%s' % where.upper():
+ return scenario_json['%s_config' % where].get('async_%s_threads' % where, 0)
+ return scenario_json['client_config']['outstanding_rpcs_per_channel'] * scenario_json['client_config']['client_channels']
-def guess_cpu(scenario_json):
- client = threads_of_type(scenario_json, 'client_config/async_client_threads')
- server = threads_of_type(scenario_json, 'server_config/async_server_threads')
+def guess_cpu(scenario_json, is_tsan):
+ client = threads_required(scenario_json, 'client', is_tsan)
+ server = threads_required(scenario_json, 'server', is_tsan)
# make an arbitrary guess if set to auto-detect
# about the size of the jenkins instances we have for unit tests
- if client == 0: client = 8
- if server == 0: server = 8
+ if client == 0 or server == 0: return 'capacity'
return (scenario_json['num_clients'] * client +
scenario_json['num_servers'] * server)
@@ -73,16 +83,33 @@ print yaml.dump({
{
'name': 'json_run_localhost',
'shortname': 'json_run_localhost:%s' % scenario_json['name'],
- 'args': ['--scenarios_json', _scenario_json_string(scenario_json)],
+ 'args': ['--scenarios_json', _scenario_json_string(scenario_json, False)],
+ 'ci_platforms': ['linux'],
+ 'platforms': ['linux'],
+ 'flaky': False,
+ 'language': 'c++',
+ 'boringssl': True,
+ 'defaults': 'boringssl',
+ 'cpu_cost': guess_cpu(scenario_json, False),
+ 'exclude_configs': ['tsan', 'asan'],
+ 'timeout_seconds': 6*60
+ }
+ for scenario_json in scenario_config.CXXLanguage().scenarios()
+ if 'scalable' in scenario_json.get('CATEGORIES', [])
+ ] + [
+ {
+ 'name': 'json_run_localhost',
+ 'shortname': 'json_run_localhost:%s_low_thread_count' % scenario_json['name'],
+ 'args': ['--scenarios_json', _scenario_json_string(scenario_json, True)],
'ci_platforms': ['linux'],
'platforms': ['linux'],
'flaky': False,
'language': 'c++',
'boringssl': True,
'defaults': 'boringssl',
- 'cpu_cost': guess_cpu(scenario_json),
- 'exclude_configs': [],
- 'timeout_seconds': 3*60
+ 'cpu_cost': guess_cpu(scenario_json, True),
+ 'exclude_configs': sorted(c for c in configs_from_yaml if c not in ('tsan', 'asan')),
+ 'timeout_seconds': 6*60
}
for scenario_json in scenario_config.CXXLanguage().scenarios()
if 'scalable' in scenario_json.get('CATEGORIES', [])
diff --git a/test/cpp/qps/interarrival.h b/test/cpp/qps/interarrival.h
index 0980d5e8ba..4bef06f566 100644
--- a/test/cpp/qps/interarrival.h
+++ b/test/cpp/qps/interarrival.h
@@ -69,11 +69,11 @@ inline RandomDistInterface::~RandomDistInterface() {}
// independent identical stationary sources. For more information,
// see http://en.wikipedia.org/wiki/Exponential_distribution
-class ExpDist GRPC_FINAL : public RandomDistInterface {
+class ExpDist final : public RandomDistInterface {
public:
explicit ExpDist(double lambda) : lambda_recip_(1.0 / lambda) {}
- ~ExpDist() GRPC_OVERRIDE {}
- double transform(double uni) const GRPC_OVERRIDE {
+ ~ExpDist() override {}
+ double transform(double uni) const override {
// Note: Use 1.0-uni above to avoid NaN if uni is 0
return lambda_recip_ * (-log(1.0 - uni));
}
diff --git a/test/cpp/qps/json_run_localhost.cc b/test/cpp/qps/json_run_localhost.cc
index 74e40fbf1a..b7b2553f12 100644
--- a/test/cpp/qps/json_run_localhost.cc
+++ b/test/cpp/qps/json_run_localhost.cc
@@ -31,7 +31,11 @@
*
*/
+#include <signal.h>
+#include <string.h>
+
#include <memory>
+#include <mutex>
#include <sstream>
#include <string>
@@ -43,6 +47,11 @@
using grpc::SubProcess;
+constexpr auto kNumWorkers = 2;
+
+static SubProcess* g_driver;
+static SubProcess* g_workers[kNumWorkers];
+
template <class T>
std::string as_string(const T& val) {
std::ostringstream out;
@@ -50,9 +59,38 @@ std::string as_string(const T& val) {
return out.str();
}
+static void sighandler(int sig) {
+ const int errno_saved = errno;
+ if (g_driver != NULL) g_driver->Interrupt();
+ for (int i = 0; i < kNumWorkers; ++i) {
+ if (g_workers[i]) g_workers[i]->Interrupt();
+ }
+ errno = errno_saved;
+}
+
+static void register_sighandler() {
+ struct sigaction act;
+ memset(&act, 0, sizeof(act));
+ act.sa_handler = sighandler;
+
+ sigaction(SIGINT, &act, NULL);
+ sigaction(SIGTERM, &act, NULL);
+}
+
+static void LogStatus(int status, const char* label) {
+ if (WIFEXITED(status)) {
+ gpr_log(GPR_INFO, "%s: subprocess exited with status %d", label,
+ WEXITSTATUS(status));
+ } else if (WIFSIGNALED(status)) {
+ gpr_log(GPR_INFO, "%s: subprocess terminated with signal %d", label,
+ WTERMSIG(status));
+ } else {
+ gpr_log(GPR_INFO, "%s: unknown subprocess status: %d", label, status);
+ }
+}
+
int main(int argc, char** argv) {
- typedef std::unique_ptr<SubProcess> SubProcessPtr;
- std::vector<SubProcessPtr> jobs;
+ register_sighandler();
std::string my_bin = argv[0];
std::string bin_dir = my_bin.substr(0, my_bin.rfind('/'));
@@ -60,11 +98,11 @@ int main(int argc, char** argv) {
std::ostringstream env;
bool first = true;
- for (int i = 0; i < 2; i++) {
- auto port = grpc_pick_unused_port_or_die();
+ for (int i = 0; i < kNumWorkers; i++) {
+ const auto port = grpc_pick_unused_port_or_die();
std::vector<std::string> args = {bin_dir + "/qps_worker", "-driver_port",
as_string(port)};
- jobs.emplace_back(new SubProcess(args));
+ g_workers[i] = new SubProcess(args);
if (!first) env << ",";
env << "localhost:" << port;
first = false;
@@ -75,12 +113,27 @@ int main(int argc, char** argv) {
for (int i = 1; i < argc; i++) {
args.push_back(argv[i]);
}
- GPR_ASSERT(SubProcess(args).Join() == 0);
- for (auto it = jobs.begin(); it != jobs.end(); ++it) {
- (*it)->Interrupt();
+ g_driver = new SubProcess(args);
+ const int driver_join_status = g_driver->Join();
+ if (driver_join_status != 0) {
+ LogStatus(driver_join_status, "driver");
}
- for (auto it = jobs.begin(); it != jobs.end(); ++it) {
- (*it)->Join();
+ for (int i = 0; i < kNumWorkers; ++i) {
+ if (g_workers[i]) g_workers[i]->Interrupt();
}
+
+ for (int i = 0; i < kNumWorkers; ++i) {
+ if (g_workers[i]) {
+ const int worker_status = g_workers[i]->Join();
+ if (worker_status != 0) {
+ LogStatus(worker_status, "worker");
+ }
+ }
+ }
+
+ delete g_driver;
+ g_driver = NULL;
+ for (int i = 0; i < kNumWorkers; ++i) delete g_workers[i];
+ GPR_ASSERT(driver_join_status == 0);
}
diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc
index 1524ebbc38..da835b995a 100644
--- a/test/cpp/qps/qps_json_driver.cc
+++ b/test/cpp/qps/qps_json_driver.cc
@@ -49,10 +49,119 @@ DEFINE_string(scenarios_file, "",
DEFINE_string(scenarios_json, "",
"JSON string containing an array of Scenario objects");
DEFINE_bool(quit, false, "Quit the workers");
+DEFINE_string(search_param, "",
+ "The parameter, whose value is to be searched for to achieve "
+ "targeted cpu load. For now, we have 'offered_load'. Later, "
+ "'num_channels', 'num_outstanding_requests', etc. shall be "
+ "added.");
+DEFINE_double(
+ initial_search_value, 0.0,
+ "initial parameter value to start the search with (i.e. lower bound)");
+DEFINE_double(targeted_cpu_load, 70.0,
+ "Targeted cpu load (unit: %, range [0,100])");
+DEFINE_double(stride, 1,
+ "Defines each stride of the search. The larger the stride is, "
+ "the coarser the result will be, but will also be faster.");
+DEFINE_double(error_tolerance, 0.01,
+ "Defines threshold for stopping the search. When current search "
+ "range is narrower than the error_tolerance computed range, we "
+ "stop the search.");
+
+DEFINE_string(qps_server_target_override, "",
+ "Override QPS server target to configure in client configs."
+ "Only applicable if there is a single benchmark server.");
+DEFINE_bool(configure_core_lists, true,
+ "Provide 'core_list' parameters to workers. Value determined "
+ "by cores available and 'core_limit' parameters of the scenarios.");
namespace grpc {
namespace testing {
+static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario,
+ bool* success) {
+ std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
+ auto result = RunScenario(
+ scenario.client_config(), scenario.num_clients(),
+ scenario.server_config(), scenario.num_servers(),
+ scenario.warmup_seconds(), scenario.benchmark_seconds(),
+ scenario.spawn_local_worker_count(),
+ FLAGS_qps_server_target_override.c_str(), FLAGS_configure_core_lists);
+
+ // Amend the result with scenario config. Eventually we should adjust
+ // RunScenario contract so we don't need to touch the result here.
+ result->mutable_scenario()->CopyFrom(scenario);
+
+ GetReporter()->ReportQPS(*result);
+ GetReporter()->ReportQPSPerCore(*result);
+ GetReporter()->ReportLatency(*result);
+ GetReporter()->ReportTimes(*result);
+ GetReporter()->ReportCpuUsage(*result);
+
+ for (int i = 0; *success && i < result->client_success_size(); i++) {
+ *success = result->client_success(i);
+ }
+ for (int i = 0; *success && i < result->server_success_size(); i++) {
+ *success = result->server_success(i);
+ }
+
+ return result;
+}
+
+static double GetCpuLoad(Scenario* scenario, double offered_load,
+ bool* success) {
+ scenario->mutable_client_config()
+ ->mutable_load_params()
+ ->mutable_poisson()
+ ->set_offered_load(offered_load);
+ auto result = RunAndReport(*scenario, success);
+ return result->summary().server_cpu_usage();
+}
+
+static double BinarySearch(Scenario* scenario, double targeted_cpu_load,
+ double low, double high, bool* success) {
+ while (low <= high * (1 - FLAGS_error_tolerance)) {
+ double mid = low + (high - low) / 2;
+ double current_cpu_load = GetCpuLoad(scenario, mid, success);
+ gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f", mid);
+ if (!*success) {
+ gpr_log(GPR_ERROR, "Client/Server Failure");
+ break;
+ }
+ if (targeted_cpu_load <= current_cpu_load) {
+ high = mid - FLAGS_stride;
+ } else {
+ low = mid + FLAGS_stride;
+ }
+ }
+
+ return low;
+}
+
+static double SearchOfferedLoad(double initial_offered_load,
+ double targeted_cpu_load, Scenario* scenario,
+ bool* success) {
+ std::cerr << "RUNNING SCENARIO: " << scenario->name() << "\n";
+ double current_offered_load = initial_offered_load;
+ double current_cpu_load = GetCpuLoad(scenario, current_offered_load, success);
+ if (current_cpu_load > targeted_cpu_load) {
+ gpr_log(GPR_ERROR, "Initial offered load too high");
+ return -1;
+ }
+
+ while (*success && (current_cpu_load < targeted_cpu_load)) {
+ current_offered_load *= 2;
+ current_cpu_load = GetCpuLoad(scenario, current_offered_load, success);
+ gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f",
+ current_offered_load);
+ }
+
+ double targeted_offered_load =
+ BinarySearch(scenario, targeted_cpu_load, current_offered_load / 2,
+ current_offered_load, success);
+
+ return targeted_offered_load;
+}
+
static bool QpsDriver() {
grpc::string json;
@@ -68,11 +177,11 @@ static bool QpsDriver() {
if (scfile) {
// Read the json data from disk
- FILE *json_file = fopen(FLAGS_scenarios_file.c_str(), "r");
+ FILE* json_file = fopen(FLAGS_scenarios_file.c_str(), "r");
GPR_ASSERT(json_file != NULL);
fseek(json_file, 0, SEEK_END);
long len = ftell(json_file);
- char *data = new char[len];
+ char* data = new char[len];
fseek(json_file, 0, SEEK_SET);
GPR_ASSERT(len == (long)fread(data, 1, len, json_file));
fclose(json_file);
@@ -93,28 +202,19 @@ static bool QpsDriver() {
GPR_ASSERT(scenarios.scenarios_size() > 0);
for (int i = 0; i < scenarios.scenarios_size(); i++) {
- const Scenario &scenario = scenarios.scenarios(i);
- std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
- auto result =
- RunScenario(scenario.client_config(), scenario.num_clients(),
- scenario.server_config(), scenario.num_servers(),
- scenario.warmup_seconds(), scenario.benchmark_seconds(),
- scenario.spawn_local_worker_count());
-
- // Amend the result with scenario config. Eventually we should adjust
- // RunScenario contract so we don't need to touch the result here.
- result->mutable_scenario()->CopyFrom(scenario);
-
- GetReporter()->ReportQPS(*result);
- GetReporter()->ReportQPSPerCore(*result);
- GetReporter()->ReportLatency(*result);
- GetReporter()->ReportTimes(*result);
-
- for (int i = 0; success && i < result->client_success_size(); i++) {
- success = result->client_success(i);
- }
- for (int i = 0; success && i < result->server_success_size(); i++) {
- success = result->server_success(i);
+ if (FLAGS_search_param == "") {
+ const Scenario& scenario = scenarios.scenarios(i);
+ RunAndReport(scenario, &success);
+ } else {
+ if (FLAGS_search_param == "offered_load") {
+ Scenario* scenario = scenarios.mutable_scenarios(i);
+ double targeted_offered_load =
+ SearchOfferedLoad(FLAGS_initial_search_value,
+ FLAGS_targeted_cpu_load, scenario, &success);
+ gpr_log(GPR_INFO, "targeted_offered_load %f", targeted_offered_load);
+ } else {
+ gpr_log(GPR_ERROR, "Unimplemented search param");
+ }
}
}
return success;
@@ -123,7 +223,7 @@ static bool QpsDriver() {
} // namespace testing
} // namespace grpc
-int main(int argc, char **argv) {
+int main(int argc, char** argv) {
grpc::testing::InitBenchmark(&argc, &argv, true);
bool ok = grpc::testing::QpsDriver();
diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc
index d3e53fe14a..d437920e68 100644
--- a/test/cpp/qps/qps_worker.cc
+++ b/test/cpp/qps/qps_worker.cc
@@ -100,7 +100,7 @@ static std::unique_ptr<Server> CreateServer(const ServerConfig& config) {
abort();
}
-class ScopedProfile GRPC_FINAL {
+class ScopedProfile final {
public:
ScopedProfile(const char* filename, bool enable) : enable_(enable) {
if (enable_) grpc_profiler_start(filename);
@@ -113,14 +113,14 @@ class ScopedProfile GRPC_FINAL {
const bool enable_;
};
-class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
+class WorkerServiceImpl final : public WorkerService::Service {
public:
WorkerServiceImpl(int server_port, QpsWorker* worker)
: acquired_(false), server_port_(server_port), worker_(worker) {}
- Status RunClient(ServerContext* ctx,
- ServerReaderWriter<ClientStatus, ClientArgs>* stream)
- GRPC_OVERRIDE {
+ Status RunClient(
+ ServerContext* ctx,
+ ServerReaderWriter<ClientStatus, ClientArgs>* stream) override {
InstanceGuard g(this);
if (!g.Acquired()) {
return Status(StatusCode::RESOURCE_EXHAUSTED, "Client worker busy");
@@ -132,9 +132,9 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
return ret;
}
- Status RunServer(ServerContext* ctx,
- ServerReaderWriter<ServerStatus, ServerArgs>* stream)
- GRPC_OVERRIDE {
+ Status RunServer(
+ ServerContext* ctx,
+ ServerReaderWriter<ServerStatus, ServerArgs>* stream) override {
InstanceGuard g(this);
if (!g.Acquired()) {
return Status(StatusCode::RESOURCE_EXHAUSTED, "Server worker busy");
@@ -147,12 +147,12 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
}
Status CoreCount(ServerContext* ctx, const CoreRequest*,
- CoreResponse* resp) GRPC_OVERRIDE {
+ CoreResponse* resp) override {
resp->set_cores(gpr_cpu_num_cores());
return Status::OK;
}
- Status QuitWorker(ServerContext* ctx, const Void*, Void*) GRPC_OVERRIDE {
+ Status QuitWorker(ServerContext* ctx, const Void*, Void*) override {
InstanceGuard g(this);
if (!g.Acquired()) {
return Status(StatusCode::RESOURCE_EXHAUSTED, "Quitting worker busy");
diff --git a/test/cpp/qps/report.cc b/test/cpp/qps/report.cc
index 41617e968a..7f84816421 100644
--- a/test/cpp/qps/report.cc
+++ b/test/cpp/qps/report.cc
@@ -71,6 +71,12 @@ void CompositeReporter::ReportTimes(const ScenarioResult& result) {
}
}
+void CompositeReporter::ReportCpuUsage(const ScenarioResult& result) {
+ for (size_t i = 0; i < reporters_.size(); ++i) {
+ reporters_[i]->ReportCpuUsage(result);
+ }
+}
+
void GprLogReporter::ReportQPS(const ScenarioResult& result) {
gpr_log(GPR_INFO, "QPS: %.1f", result.summary().qps());
if (result.summary().failed_requests_per_second() > 0) {
@@ -107,6 +113,11 @@ void GprLogReporter::ReportTimes(const ScenarioResult& result) {
result.summary().client_user_time());
}
+void GprLogReporter::ReportCpuUsage(const ScenarioResult& result) {
+ gpr_log(GPR_INFO, "Server CPU usage: %.2f%%",
+ result.summary().server_cpu_usage());
+}
+
void JsonReporter::ReportQPS(const ScenarioResult& result) {
grpc::string json_string =
SerializeJson(result, "type.googleapis.com/grpc.testing.ScenarioResult");
@@ -127,5 +138,9 @@ void JsonReporter::ReportTimes(const ScenarioResult& result) {
// NOP - all reporting is handled by ReportQPS.
}
+void JsonReporter::ReportCpuUsage(const ScenarioResult& result) {
+ // NOP - all reporting is handled by ReportQPS.
+}
+
} // namespace testing
} // namespace grpc
diff --git a/test/cpp/qps/report.h b/test/cpp/qps/report.h
index 39cf498e7b..faf87ff060 100644
--- a/test/cpp/qps/report.h
+++ b/test/cpp/qps/report.h
@@ -70,6 +70,9 @@ class Reporter {
/** Reports system and user time for client and server systems. */
virtual void ReportTimes(const ScenarioResult& result) = 0;
+ /** Reports server cpu usage. */
+ virtual void ReportCpuUsage(const ScenarioResult& result) = 0;
+
private:
const string name_;
};
@@ -82,10 +85,11 @@ class CompositeReporter : public Reporter {
/** Adds a \a reporter to the composite. */
void add(std::unique_ptr<Reporter> reporter);
- void ReportQPS(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportQPSPerCore(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportLatency(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportTimes(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportQPS(const ScenarioResult& result) override;
+ void ReportQPSPerCore(const ScenarioResult& result) override;
+ void ReportLatency(const ScenarioResult& result) override;
+ void ReportTimes(const ScenarioResult& result) override;
+ void ReportCpuUsage(const ScenarioResult& result) override;
private:
std::vector<std::unique_ptr<Reporter> > reporters_;
@@ -97,10 +101,11 @@ class GprLogReporter : public Reporter {
GprLogReporter(const string& name) : Reporter(name) {}
private:
- void ReportQPS(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportQPSPerCore(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportLatency(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportTimes(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportQPS(const ScenarioResult& result) override;
+ void ReportQPSPerCore(const ScenarioResult& result) override;
+ void ReportLatency(const ScenarioResult& result) override;
+ void ReportTimes(const ScenarioResult& result) override;
+ void ReportCpuUsage(const ScenarioResult& result) override;
};
/** Dumps the report to a JSON file. */
@@ -110,10 +115,11 @@ class JsonReporter : public Reporter {
: Reporter(name), report_file_(report_file) {}
private:
- void ReportQPS(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportQPSPerCore(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportLatency(const ScenarioResult& result) GRPC_OVERRIDE;
- void ReportTimes(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportQPS(const ScenarioResult& result) override;
+ void ReportQPSPerCore(const ScenarioResult& result) override;
+ void ReportLatency(const ScenarioResult& result) override;
+ void ReportTimes(const ScenarioResult& result) override;
+ void ReportCpuUsage(const ScenarioResult& result) override;
const string report_file_;
};
diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h
index e8bc396696..c3d18e5789 100644
--- a/test/cpp/qps/server.h
+++ b/test/cpp/qps/server.h
@@ -75,6 +75,8 @@ class Server {
stats.set_time_elapsed(timer_result.wall);
stats.set_time_system(timer_result.system);
stats.set_time_user(timer_result.user);
+ stats.set_total_cpu_time(timer_result.total_cpu_time);
+ stats.set_idle_cpu_time(timer_result.idle_cpu_time);
return stats;
}
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index bc4c896d83..b3a06aeaf5 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -58,7 +58,7 @@ namespace testing {
template <class RequestType, class ResponseType, class ServiceType,
class ServerContextType>
-class AsyncQpsServerTest GRPC_FINAL : public grpc::testing::Server {
+class AsyncQpsServerTest final : public grpc::testing::Server {
public:
AsyncQpsServerTest(
const ServerConfig &config,
@@ -196,7 +196,7 @@ class AsyncQpsServerTest GRPC_FINAL : public grpc::testing::Server {
return reinterpret_cast<ServerRpcContext *>(tag);
}
- class ServerRpcContextUnaryImpl GRPC_FINAL : public ServerRpcContext {
+ class ServerRpcContextUnaryImpl final : public ServerRpcContext {
public:
ServerRpcContextUnaryImpl(
std::function<void(ServerContextType *, RequestType *,
@@ -213,11 +213,9 @@ class AsyncQpsServerTest GRPC_FINAL : public grpc::testing::Server {
request_method_(srv_ctx_.get(), &req_, &response_writer_,
AsyncQpsServerTest::tag(this));
}
- ~ServerRpcContextUnaryImpl() GRPC_OVERRIDE {}
- bool RunNextState(bool ok) GRPC_OVERRIDE {
- return (this->*next_state_)(ok);
- }
- void Reset() GRPC_OVERRIDE {
+ ~ServerRpcContextUnaryImpl() override {}
+ bool RunNextState(bool ok) override { return (this->*next_state_)(ok); }
+ void Reset() override {
srv_ctx_.reset(new ServerContextType);
req_ = RequestType();
response_writer_ =
@@ -257,7 +255,7 @@ class AsyncQpsServerTest GRPC_FINAL : public grpc::testing::Server {
grpc::ServerAsyncResponseWriter<ResponseType> response_writer_;
};
- class ServerRpcContextStreamingImpl GRPC_FINAL : public ServerRpcContext {
+ class ServerRpcContextStreamingImpl final : public ServerRpcContext {
public:
ServerRpcContextStreamingImpl(
std::function<void(
@@ -273,11 +271,9 @@ class AsyncQpsServerTest GRPC_FINAL : public grpc::testing::Server {
stream_(srv_ctx_.get()) {
request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
}
- ~ServerRpcContextStreamingImpl() GRPC_OVERRIDE {}
- bool RunNextState(bool ok) GRPC_OVERRIDE {
- return (this->*next_state_)(ok);
- }
- void Reset() GRPC_OVERRIDE {
+ ~ServerRpcContextStreamingImpl() override {}
+ bool RunNextState(bool ok) override { return (this->*next_state_)(ok); }
+ void Reset() override {
srv_ctx_.reset(new ServerContextType);
req_ = RequestType();
stream_ = grpc::ServerAsyncReaderWriter<ResponseType, RequestType>(
@@ -380,7 +376,7 @@ static Status ProcessGenericRPC(const PayloadConfig &payload_config,
ByteBuffer *response) {
int resp_size = payload_config.bytebuf_params().resp_size();
std::unique_ptr<char[]> buf(new char[resp_size]);
- gpr_slice s = gpr_slice_from_copied_buffer(buf.get(), resp_size);
+ grpc_slice s = grpc_slice_from_copied_buffer(buf.get(), resp_size);
Slice slice(s, Slice::STEAL_REF);
*response = ByteBuffer(&slice, 1);
return Status::OK;
diff --git a/test/cpp/qps/server_sync.cc b/test/cpp/qps/server_sync.cc
index 07f48e2644..8076a4a6b9 100644
--- a/test/cpp/qps/server_sync.cc
+++ b/test/cpp/qps/server_sync.cc
@@ -48,10 +48,10 @@
namespace grpc {
namespace testing {
-class BenchmarkServiceImpl GRPC_FINAL : public BenchmarkService::Service {
+class BenchmarkServiceImpl final : public BenchmarkService::Service {
public:
Status UnaryCall(ServerContext* context, const SimpleRequest* request,
- SimpleResponse* response) GRPC_OVERRIDE {
+ SimpleResponse* response) override {
if (request->response_size() > 0) {
if (!Server::SetPayload(request->response_type(),
request->response_size(),
@@ -63,7 +63,7 @@ class BenchmarkServiceImpl GRPC_FINAL : public BenchmarkService::Service {
}
Status StreamingCall(
ServerContext* context,
- ServerReaderWriter<SimpleResponse, SimpleRequest>* stream) GRPC_OVERRIDE {
+ ServerReaderWriter<SimpleResponse, SimpleRequest>* stream) override {
SimpleRequest request;
while (stream->Read(&request)) {
SimpleResponse response;
@@ -80,7 +80,7 @@ class BenchmarkServiceImpl GRPC_FINAL : public BenchmarkService::Service {
}
};
-class SynchronousServer GRPC_FINAL : public grpc::testing::Server {
+class SynchronousServer final : public grpc::testing::Server {
public:
explicit SynchronousServer(const ServerConfig& config) : Server(config) {
ServerBuilder builder;
diff --git a/test/cpp/qps/usage_timer.cc b/test/cpp/qps/usage_timer.cc
index ff595b2ba0..c6697fbdfd 100644
--- a/test/cpp/qps/usage_timer.cc
+++ b/test/cpp/qps/usage_timer.cc
@@ -33,10 +33,14 @@
#include "test/cpp/qps/usage_timer.h"
+#include <fstream>
+#include <sstream>
+#include <string>
+
+#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <sys/resource.h>
#include <sys/time.h>
-
UsageTimer::UsageTimer() : start_(Sample()) {}
double UsageTimer::Now() {
@@ -48,6 +52,27 @@ static double time_double(struct timeval* tv) {
return tv->tv_sec + 1e-6 * tv->tv_usec;
}
+static void get_cpu_usage(unsigned long long* total_cpu_time,
+ unsigned long long* idle_cpu_time) {
+#ifdef __linux__
+ std::ifstream proc_stat("/proc/stat");
+ proc_stat.ignore(5);
+ std::string cpu_time_str;
+ std::string first_line;
+ std::getline(proc_stat, first_line);
+ std::stringstream first_line_s(first_line);
+ for (int i = 0; i < 10; ++i) {
+ std::getline(first_line_s, cpu_time_str, ' ');
+ *total_cpu_time += std::stol(cpu_time_str);
+ if (i == 3) {
+ *idle_cpu_time = std::stol(cpu_time_str);
+ }
+ }
+#else
+ gpr_log(GPR_INFO, "get_cpu_usage(): Non-linux platform is not supported.");
+#endif
+}
+
UsageTimer::Result UsageTimer::Sample() {
struct rusage usage;
struct timeval tv;
@@ -58,6 +83,9 @@ UsageTimer::Result UsageTimer::Sample() {
r.wall = time_double(&tv);
r.user = time_double(&usage.ru_utime);
r.system = time_double(&usage.ru_stime);
+ r.total_cpu_time = 0;
+ r.idle_cpu_time = 0;
+ get_cpu_usage(&r.total_cpu_time, &r.idle_cpu_time);
return r;
}
@@ -67,5 +95,8 @@ UsageTimer::Result UsageTimer::Mark() const {
r.wall = s.wall - start_.wall;
r.user = s.user - start_.user;
r.system = s.system - start_.system;
+ r.total_cpu_time = s.total_cpu_time - start_.total_cpu_time;
+ r.idle_cpu_time = s.idle_cpu_time - start_.idle_cpu_time;
+
return r;
}
diff --git a/test/cpp/qps/usage_timer.h b/test/cpp/qps/usage_timer.h
index 8343cd6653..0fc1b47996 100644
--- a/test/cpp/qps/usage_timer.h
+++ b/test/cpp/qps/usage_timer.h
@@ -42,6 +42,8 @@ class UsageTimer {
double wall;
double user;
double system;
+ unsigned long long total_cpu_time;
+ unsigned long long idle_cpu_time;
};
Result Mark() const;
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-04 14:30:31 +0000