clang-format

reviewable/pr3905/r8
Vijay Pai 9 years ago
parent 780a7f205d
commit ce84670628
  1. 4
      test/cpp/qps/async_streaming_ping_pong_test.cc
  2. 4
      test/cpp/qps/async_unary_ping_pong_test.cc
  3. 50
      test/cpp/qps/client.h
  4. 17
      test/cpp/qps/client_async.cc
  5. 16
      test/cpp/qps/driver.cc
  6. 2
      test/cpp/qps/driver.h
  7. 6
      test/cpp/qps/qps_driver.cc
  8. 4
      test/cpp/qps/qps_openloop_test.cc
  9. 4
      test/cpp/qps/qps_test.cc
  10. 7
      test/cpp/qps/qps_worker.cc
  11. 10
      test/cpp/qps/server.h
  12. 15
      test/cpp/qps/server_async.cc
  13. 7
      test/cpp/qps/server_sync.cc
  14. 92
      test/cpp/qps/single_run_localhost.sh
  15. 4
      test/cpp/qps/sync_streaming_ping_pong_test.cc
  16. 4
      test/cpp/qps/sync_unary_ping_pong_test.cc
  17. 4
      test/proto/benchmarks/control.proto
  18. 5
      test/proto/benchmarks/payloads.proto
  19. 1
      test/proto/benchmarks/stats.proto

@ -54,7 +54,9 @@ static void RunAsyncStreamingPingPong() {
client_config.set_client_type(ASYNC_CLIENT);
client_config.set_outstanding_rpcs_per_channel(1);
client_config.set_client_channels(1);
client_config.mutable_payload_config()->mutable_simple_params()->set_resp_size(1);
client_config.mutable_payload_config()
->mutable_simple_params()
->set_resp_size(1);
client_config.set_async_client_threads(1);
client_config.set_rpc_type(STREAMING);
client_config.mutable_load_params()->mutable_closed_loop();

@ -54,7 +54,9 @@ static void RunAsyncUnaryPingPong() {
client_config.set_client_type(ASYNC_CLIENT);
client_config.set_outstanding_rpcs_per_channel(1);
client_config.set_client_channels(1);
client_config.mutable_payload_config()->mutable_simple_params()->set_resp_size(1);
client_config.mutable_payload_config()
->mutable_simple_params()
->set_resp_size(1);
client_config.set_async_client_threads(1);
client_config.set_rpc_type(UNARY);
client_config.mutable_load_params()->mutable_closed_loop();

@ -77,18 +77,20 @@ class Client {
config);
}
if (config.payload_config().has_bytebuf_params()) {
GPR_ASSERT(false); // not yet implemented
GPR_ASSERT(false); // not yet implemented
} else if (config.payload_config().has_simple_params()) {
request_.set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
request_.set_response_size(config.payload_config().simple_params().resp_size());
request_.mutable_payload()->set_type(grpc::testing::PayloadType::COMPRESSABLE);
request_.set_response_size(
config.payload_config().simple_params().resp_size());
request_.mutable_payload()->set_type(
grpc::testing::PayloadType::COMPRESSABLE);
int size = config.payload_config().simple_params().req_size();
std::unique_ptr<char[]> body(new char[size]);
request_.mutable_payload()->set_body(body.get(), size);
} else if (config.payload_config().has_complex_params()) {
GPR_ASSERT(false); // not yet implemented
GPR_ASSERT(false); // not yet implemented
} else {
GPR_ASSERT(false); // badly configured
GPR_ASSERT(false); // badly configured
}
}
virtual ~Client() {}
@ -101,20 +103,20 @@ class Client {
if (reset) {
Histogram* to_merge = new Histogram[threads_.size()];
for (size_t i = 0; i < threads_.size(); i++) {
threads_[i]->BeginSwap(&to_merge[i]);
threads_[i]->BeginSwap(&to_merge[i]);
}
std::unique_ptr<Timer> timer(new Timer);
timer_.swap(timer);
for (size_t i = 0; i < threads_.size(); i++) {
threads_[i]->EndSwap();
latencies.Merge(to_merge[i]);
threads_[i]->EndSwap();
latencies.Merge(to_merge[i]);
}
delete[] to_merge;
timer_result = timer->Mark();
} else {
// merge snapshots of each thread histogram
for (size_t i = 0; i < threads_.size(); i++) {
threads_[i]->MergeStatsInto(&latencies);
threads_[i]->MergeStatsInto(&latencies);
}
timer_result = timer_->Mark();
}
@ -144,11 +146,10 @@ class Client {
// We have to use a 2-phase init like this with a default
// constructor followed by an initializer function to make
// old compilers happy with using this in std::vector
channel_ =
CreateTestChannel(target,
config.security_params().server_host_override(),
config.has_security_params(),
!config.security_params().use_test_ca());
channel_ = CreateTestChannel(
target, config.security_params().server_host_override(),
config.has_security_params(),
!config.security_params().use_test_ca());
stub_ = BenchmarkService::NewStub(channel_);
}
Channel* get_channel() { return channel_.get(); }
@ -176,21 +177,22 @@ class Client {
std::unique_ptr<RandomDist> random_dist;
if (load.has_poisson()) {
random_dist.reset(new ExpDist(load.poisson().offered_load() /
num_threads));
random_dist.reset(
new ExpDist(load.poisson().offered_load() / num_threads));
} else if (load.has_uniform()) {
random_dist.reset(new UniformDist(load.uniform().interarrival_lo() *
num_threads,
load.uniform().interarrival_hi() *
num_threads));
random_dist.reset(
new UniformDist(load.uniform().interarrival_lo() * num_threads,
load.uniform().interarrival_hi() * num_threads));
} else if (load.has_determ()) {
random_dist.reset(new DetDist(num_threads / load.determ().offered_load()));
random_dist.reset(
new DetDist(num_threads / load.determ().offered_load()));
} else if (load.has_pareto()) {
random_dist.reset(new ParetoDist(load.pareto().interarrival_base() * num_threads,
load.pareto().alpha()));
random_dist.reset(
new ParetoDist(load.pareto().interarrival_base() * num_threads,
load.pareto().alpha()));
} else if (load.has_closed_loop()) {
// Closed-loop doesn't use random dist at all
} else { // invalid load type
} else { // invalid load type
GPR_ASSERT(false);
}

@ -358,7 +358,8 @@ class AsyncUnaryClient GRPC_FINAL : public AsyncClient {
const SimpleRequest& request, CompletionQueue* cq) {
return stub->AsyncUnaryCall(ctx, request, cq);
};
static ClientRpcContext* SetupCtx(int channel_id, BenchmarkService::Stub* stub,
static ClientRpcContext* SetupCtx(int channel_id,
BenchmarkService::Stub* stub,
const SimpleRequest& req) {
return new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
channel_id, stub, req, AsyncUnaryClient::StartReq,
@ -371,9 +372,10 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
public:
ClientRpcContextStreamingImpl(
int channel_id, BenchmarkService::Stub* stub, const RequestType& req,
std::function<std::unique_ptr<grpc::ClientAsyncReaderWriter<
RequestType, ResponseType>>(BenchmarkService::Stub*, grpc::ClientContext*,
CompletionQueue*, void*)> start_req,
std::function<std::unique_ptr<
grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*,
void*)> start_req,
std::function<void(grpc::Status, ResponseType*)> on_done)
: ClientRpcContext(channel_id),
context_(),
@ -427,8 +429,8 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
std::function<void(grpc::Status, ResponseType*)> callback_;
std::function<
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*, void*)>
start_req_;
BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*,
void*)> start_req_;
grpc::Status status_;
double start_;
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>
@ -456,7 +458,8 @@ class AsyncStreamingClient GRPC_FINAL : public AsyncClient {
auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
return stream;
};
static ClientRpcContext* SetupCtx(int channel_id, BenchmarkService::Stub* stub,
static ClientRpcContext* SetupCtx(int channel_id,
BenchmarkService::Stub* stub,
const SimpleRequest& req) {
return new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
channel_id, stub, req, AsyncStreamingClient::StartReq,

@ -161,8 +161,8 @@ std::unique_ptr<ScenarioResult> RunScenario(
// where class contained in std::vector must have a copy constructor
auto* servers = new ServerData[num_servers];
for (size_t i = 0; i < num_servers; i++) {
servers[i].stub =
WorkerService::NewStub(CreateChannel(workers[i], InsecureCredentials()));
servers[i].stub = WorkerService::NewStub(
CreateChannel(workers[i], InsecureCredentials()));
ServerArgs args;
result_server_config = server_config;
*args.mutable_setup() = server_config;
@ -248,18 +248,16 @@ std::unique_ptr<ScenarioResult> RunScenario(
for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
GPR_ASSERT(server->stream->Read(&server_status));
const auto& stats = server_status.stats();
result->server_resources.emplace_back(stats.time_elapsed(),
stats.time_user(),
stats.time_system(),
server_status.cores());
result->server_resources.emplace_back(
stats.time_elapsed(), stats.time_user(), stats.time_system(),
server_status.cores());
}
for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
GPR_ASSERT(client->stream->Read(&client_status));
const auto& stats = client_status.stats();
result->latencies.MergeProto(stats.latencies());
result->client_resources.emplace_back(stats.time_elapsed(),
stats.time_user(),
stats.time_system(), -1);
result->client_resources.emplace_back(
stats.time_elapsed(), stats.time_user(), stats.time_system(), -1);
}
for (auto client = &clients[0]; client != &clients[num_clients]; client++) {

@ -44,7 +44,7 @@ namespace testing {
class ResourceUsage {
public:
ResourceUsage(double w, double u, double s, int c)
: wall_time_(w), user_time_(u), system_time_(s), cores_(c) {}
: wall_time_(w), user_time_(u), system_time_(s), cores_(c) {}
double wall_time() const { return wall_time_; }
double user_time() const { return user_time_; }
double system_time() const { return system_time_; }

@ -101,16 +101,16 @@ static void QpsDriver() {
// Decide which type to use based on the response type
if (FLAGS_simple_resp_size >= 0) {
auto params = client_config.mutable_payload_config()->mutable_simple_params();
auto params =
client_config.mutable_payload_config()->mutable_simple_params();
params->set_resp_size(FLAGS_simple_resp_size);
if (FLAGS_simple_req_size >= 0) {
params->set_req_size(FLAGS_simple_req_size);
}
} else {
GPR_ASSERT(false); // not yet implemented
GPR_ASSERT(false); // not yet implemented
}
client_config.set_async_client_threads(FLAGS_async_client_threads);
client_config.set_rpc_type(rpc_type);

@ -54,7 +54,9 @@ static void RunQPS() {
client_config.set_client_type(ASYNC_CLIENT);
client_config.set_outstanding_rpcs_per_channel(1000);
client_config.set_client_channels(8);
client_config.mutable_payload_config()->mutable_simple_params()->set_resp_size(1);
client_config.mutable_payload_config()
->mutable_simple_params()
->set_resp_size(1);
client_config.set_async_client_threads(8);
client_config.set_rpc_type(UNARY);
client_config.mutable_load_params()->mutable_poisson()->set_offered_load(

@ -54,7 +54,9 @@ static void RunQPS() {
client_config.set_client_type(ASYNC_CLIENT);
client_config.set_outstanding_rpcs_per_channel(1000);
client_config.set_client_channels(8);
client_config.mutable_payload_config()->mutable_simple_params()->set_resp_size(1);
client_config.mutable_payload_config()
->mutable_simple_params()
->set_resp_size(1);
client_config.set_async_client_threads(8);
client_config.set_rpc_type(UNARY);
client_config.mutable_load_params()->mutable_closed_loop();

@ -76,8 +76,7 @@ static std::unique_ptr<Client> CreateClient(const ClientConfig& config) {
abort();
}
static void LimitCores(int cores) {
}
static void LimitCores(int cores) {}
static std::unique_ptr<Server> CreateServer(const ServerConfig& config) {
if (config.core_limit() > 0) {
@ -99,7 +98,7 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
explicit WorkerServiceImpl() : acquired_(false) {}
Status RunClient(ServerContext* ctx,
ServerReaderWriter<ClientStatus, ClientArgs>* stream)
ServerReaderWriter<ClientStatus, ClientArgs>* stream)
GRPC_OVERRIDE {
InstanceGuard g(this);
if (!g.Acquired()) {
@ -159,7 +158,7 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
}
Status RunClientBody(ServerContext* ctx,
ServerReaderWriter<ClientStatus, ClientArgs>* stream) {
ServerReaderWriter<ClientStatus, ClientArgs>* stream) {
ClientArgs args;
if (!stream->Read(&args)) {
return Status(StatusCode::INVALID_ARGUMENT, "");

@ -86,12 +86,13 @@ class Server {
return true;
}
int Port() const {return port_;}
int Cores() const {return gpr_cpu_num_cores();}
static std::shared_ptr<ServerCredentials> CreateServerCredentials(const ServerConfig &config) {
int Port() const { return port_; }
int Cores() const { return gpr_cpu_num_cores(); }
static std::shared_ptr<ServerCredentials> CreateServerCredentials(
const ServerConfig& config) {
if (config.has_security_params()) {
SslServerCredentialsOptions::PemKeyCertPair pkcp = {test_server1_key,
test_server1_cert};
test_server1_cert};
SslServerCredentialsOptions ssl_opts;
ssl_opts.pem_root_certs = "";
ssl_opts.pem_key_cert_pairs.push_back(pkcp);
@ -100,6 +101,7 @@ class Server {
return InsecureServerCredentials();
}
}
private:
int port_;
std::unique_ptr<Timer> timer_;

@ -57,13 +57,14 @@ namespace testing {
class AsyncQpsServerTest : public Server {
public:
explicit AsyncQpsServerTest(const ServerConfig &config): Server(config) {
explicit AsyncQpsServerTest(const ServerConfig &config) : Server(config) {
char *server_address = NULL;
gpr_join_host_port(&server_address, "::", Port());
ServerBuilder builder;
builder.AddListeningPort(server_address, Server::CreateServerCredentials(config));
builder.AddListeningPort(server_address,
Server::CreateServerCredentials(config));
gpr_free(server_address);
builder.RegisterAsyncService(&async_service_);
@ -77,11 +78,11 @@ class AsyncQpsServerTest : public Server {
for (int i = 0; i < 10000 / config.async_server_threads(); i++) {
for (int j = 0; j < config.async_server_threads(); j++) {
auto request_unary = std::bind(
&BenchmarkService::AsyncService::RequestUnaryCall, &async_service_, _1,
_2, _3, srv_cqs_[j].get(), srv_cqs_[j].get(), _4);
&BenchmarkService::AsyncService::RequestUnaryCall, &async_service_,
_1, _2, _3, srv_cqs_[j].get(), srv_cqs_[j].get(), _4);
auto request_streaming = std::bind(
&BenchmarkService::AsyncService::RequestStreamingCall, &async_service_,
_1, _2, srv_cqs_[j].get(), srv_cqs_[j].get(), _3);
&BenchmarkService::AsyncService::RequestStreamingCall,
&async_service_, _1, _2, srv_cqs_[j].get(), srv_cqs_[j].get(), _3);
contexts_.push_front(
new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
request_unary, ProcessRPC));
@ -334,7 +335,7 @@ class AsyncQpsServerTest : public Server {
std::vector<std::unique_ptr<PerThreadShutdownState>> shutdown_state_;
};
std::unique_ptr<Server> CreateAsyncServer(const ServerConfig& config) {
std::unique_ptr<Server> CreateAsyncServer(const ServerConfig &config) {
return std::unique_ptr<Server>(new AsyncQpsServerTest(config));
}

@ -84,20 +84,21 @@ class BenchmarkServiceImpl GRPC_FINAL : public BenchmarkService::Service {
class SynchronousServer GRPC_FINAL : public grpc::testing::Server {
public:
explicit SynchronousServer(const ServerConfig& config)
: Server(config) {
explicit SynchronousServer(const ServerConfig& config) : Server(config) {
ServerBuilder builder;
char* server_address = NULL;
gpr_join_host_port(&server_address, "::", Port());
builder.AddListeningPort(server_address, Server::CreateServerCredentials(config));
builder.AddListeningPort(server_address,
Server::CreateServerCredentials(config));
gpr_free(server_address);
builder.RegisterService(&service_);
impl_ = builder.BuildAndStart();
}
private:
BenchmarkServiceImpl service_;
std::unique_ptr<grpc::Server> impl_;

@ -1,56 +1,64 @@
#!/bin/sh
# Copyright 2015, Google Inc.
# All rights reserved.
#!/ bin / sh
#Copyright 2015, Google Inc.
#All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#Redistribution and use in source and binary forms, with or without
#modification, are permitted provided that the following conditions are
#met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#* Redistributions of source code must retain the above copyright
#notice, this list of conditions and the following disclaimer.
#* Redistributions in binary form must reproduce the above
#copyright notice, this list of conditions and the following disclaimer
#in the documentation and / or other materials provided with the
#distribution.
#* Neither the name of Google Inc.nor the names of its
#contributors may be used to endorse or promote products derived from
#this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
#"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
#LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
#A PARTICULAR PURPOSE ARE DISCLAIMED.IN NO EVENT SHALL THE COPYRIGHT
#OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
#SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES(INCLUDING, BUT NOT
#LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
#DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
#THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
#(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
#OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
# performs a single qps run with one client and one server
#performs a single qps run with one client and one server
set -ex
set -
ex
cd $(dirname $0)/../../..
cd
$(dirname $0) /
../../
..
killall qps_worker || true
killall qps_worker
|| true
config=opt
config = opt
NUMCPUS=`python2.7 -c 'import multiprocessing; print multiprocessing.cpu_count()'`
NUMCPUS =`python2 .7 - c
'import multiprocessing; print multiprocessing.cpu_count()'`
make CONFIG=$config qps_worker qps_driver -j$NUMCPUS
make CONFIG = $config qps_worker qps_driver -
j$NUMCPUS
bins/$config/qps_worker -driver_port 10000 &
PID1=$!
bins/$config/qps_worker -driver_port 10010 &
PID2=$!
bins
/ $config / qps_worker
- driver_port
10000 &PID1 = $ !bins / $config / qps_worker
- driver_port 10010 &PID2 = $ !
export QPS_WORKERS="localhost:10000,localhost:10010"
export QPS_WORKERS = "localhost:10000,localhost:10010"
bins/$config/qps_driver $*
kill -2 $PID1 $PID2
wait
bins
/ $config / qps_driver $ *
kill
- 2 $PID1 $PID2 wait

@ -54,7 +54,9 @@ static void RunSynchronousStreamingPingPong() {
client_config.set_client_type(SYNC_CLIENT);
client_config.set_outstanding_rpcs_per_channel(1);
client_config.set_client_channels(1);
client_config.mutable_payload_config()->mutable_simple_params()->set_resp_size(1);
client_config.mutable_payload_config()
->mutable_simple_params()
->set_resp_size(1);
client_config.set_rpc_type(STREAMING);
client_config.mutable_load_params()->mutable_closed_loop();

@ -54,7 +54,9 @@ static void RunSynchronousUnaryPingPong() {
client_config.set_client_type(SYNC_CLIENT);
client_config.set_outstanding_rpcs_per_channel(1);
client_config.set_client_channels(1);
client_config.mutable_payload_config()->mutable_simple_params()->set_resp_size(1);
client_config.mutable_payload_config()
->mutable_simple_params()
->set_resp_size(1);
client_config.set_rpc_type(UNARY);
client_config.mutable_load_params()->mutable_closed_loop();

@ -82,8 +82,8 @@ message LoadParams {
// presence of SecurityParams implies use of TLS
message SecurityParams {
bool use_test_ca = 1;
string server_host_override = 2;
bool use_test_ca = 1;
string server_host_override = 2;
}
message ClientConfig {

@ -42,8 +42,8 @@ message SimpleProtoParams {
}
message ComplexProtoParams {
// TODO (vpai): Fill this in once the details of complex, representative
// protos are decided
// TODO (vpai): Fill this in once the details of complex, representative
// protos are decided
}
message PayloadConfig {
@ -53,4 +53,3 @@ message PayloadConfig {
ComplexProtoParams complex_params = 3;
}
}

@ -57,4 +57,3 @@ message ClientStats {
double time_user = 3;
double time_system = 4;
}

Loading…
Cancel
Save