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

Merge pull request #1703 from ctiller/fugly

Browse Source 
clang-format qps code - it was getting out of hand
pull/1725/head
Vijay Pai 10 years ago
parent fdc9af8dca 5c8737d171
commit a4c44c2bcf
10 changed files with 170 additions and 155 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. 11
      test/cpp/qps/client.h
  2. 81
      test/cpp/qps/client_async.cc
  3. 8
      test/cpp/qps/client_sync.cc
  4. 5
      test/cpp/qps/driver.cc
  5. 4
      test/cpp/qps/qps_driver.cc
  6. 24
      test/cpp/qps/qps_worker.cc
  7. 12
      test/cpp/qps/report.cc
  8. 78
      test/cpp/qps/server_async.cc
  9. 6
      test/cpp/qps/server_sync.cc
  10. 2
      test/cpp/qps/worker.cc

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

@ -122,7 +122,9 @@ class Client {
gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
done_ = true;
}
if (done_) {return;}
if (done_) {
return;
}
// check if we're marking, swap out the histogram if so
if (new_) {
new_->Swap(&histogram_);
@ -168,10 +170,9 @@ class Client {
std::unique_ptr<Timer> timer_;
};
std::unique_ptr<Client>
CreateSynchronousUnaryClient(const ClientConfig& args);
std::unique_ptr<Client>
CreateSynchronousStreamingClient(const ClientConfig& args);
std::unique_ptr<Client> CreateSynchronousUnaryClient(const ClientConfig& args);
std::unique_ptr<Client> CreateSynchronousStreamingClient(
const ClientConfig& args);
std::unique_ptr<Client> CreateAsyncUnaryClient(const ClientConfig& args);
std::unique_ptr<Client> CreateAsyncStreamingClient(const ClientConfig& args);

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

@ -129,8 +129,8 @@ class AsyncClient : public Client {
public:
explicit AsyncClient(const ClientConfig& config,
std::function<void(CompletionQueue*, TestService::Stub*,
const SimpleRequest&)> setup_ctx) :
Client(config) {
const SimpleRequest&)> setup_ctx)
: Client(config) {
for (int i = 0; i < config.async_client_threads(); i++) {
cli_cqs_.emplace_back(new CompletionQueue);
}
@ -155,16 +155,19 @@ class AsyncClient : public Client {
}
}
bool ThreadFunc(Histogram* histogram, size_t thread_idx)
GRPC_OVERRIDE GRPC_FINAL {
bool ThreadFunc(Histogram* histogram,
size_t thread_idx) GRPC_OVERRIDE GRPC_FINAL {
void* got_tag;
bool ok;
switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok,
std::chrono::system_clock::now() +
std::chrono::seconds(1))) {
case CompletionQueue::SHUTDOWN: return false;
case CompletionQueue::TIMEOUT: return true;
case CompletionQueue::GOT_EVENT: break;
switch (cli_cqs_[thread_idx]->AsyncNext(
&got_tag, &ok,
std::chrono::system_clock::now() + std::chrono::seconds(1))) {
case CompletionQueue::SHUTDOWN:
return false;
case CompletionQueue::TIMEOUT:
return true;
case CompletionQueue::GOT_EVENT:
break;
}
ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
@ -177,18 +180,20 @@ class AsyncClient : public Client {
return true;
}
private:
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
};
class AsyncUnaryClient GRPC_FINAL : public AsyncClient {
public:
explicit AsyncUnaryClient(const ClientConfig& config) :
AsyncClient(config, SetupCtx) {
explicit AsyncUnaryClient(const ClientConfig& config)
: AsyncClient(config, SetupCtx) {
StartThreads(config.async_client_threads());
}
~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); }
private:
private:
static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub,
const SimpleRequest& req) {
auto check_done = [](grpc::Status s, SimpleResponse* response) {};
@ -205,12 +210,11 @@ template <class RequestType, class ResponseType>
class ClientRpcContextStreamingImpl : public ClientRpcContext {
public:
ClientRpcContextStreamingImpl(
TestService::Stub *stub, const RequestType &req,
std::function<
std::unique_ptr<grpc::ClientAsyncReaderWriter<
RequestType,ResponseType>>(
TestService::Stub *, grpc::ClientContext *, void *)> start_req,
std::function<void(grpc::Status, ResponseType *)> on_done)
TestService::Stub* stub, const RequestType& req,
std::function<std::unique_ptr<
grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
TestService::Stub*, grpc::ClientContext*, void*)> start_req,
std::function<void(grpc::Status, ResponseType*)> on_done)
: context_(),
stub_(stub),
req_(req),
@ -221,7 +225,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
start_(Timer::Now()),
stream_(start_req_(stub_, &context_, ClientRpcContext::tag(this))) {}
~ClientRpcContextStreamingImpl() GRPC_OVERRIDE {}
bool RunNextState(bool ok, Histogram *hist) GRPC_OVERRIDE {
bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE {
return (this->*next_state_)(ok, hist);
}
void StartNewClone() GRPC_OVERRIDE {
@ -229,59 +233,58 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
}
private:
bool ReqSent(bool ok, Histogram *) {
return StartWrite(ok);
}
bool ReqSent(bool ok, Histogram*) { return StartWrite(ok); }
bool StartWrite(bool ok) {
if (!ok) {
return(false);
return (false);
}
start_ = Timer::Now();
next_state_ = &ClientRpcContextStreamingImpl::WriteDone;
stream_->Write(req_, ClientRpcContext::tag(this));
return true;
}
bool WriteDone(bool ok, Histogram *) {
bool WriteDone(bool ok, Histogram*) {
if (!ok) {
return(false);
return (false);
}
next_state_ = &ClientRpcContextStreamingImpl::ReadDone;
stream_->Read(&response_, ClientRpcContext::tag(this));
return true;
}
bool ReadDone(bool ok, Histogram *hist) {
bool ReadDone(bool ok, Histogram* hist) {
hist->Add((Timer::Now() - start_) * 1e9);
return StartWrite(ok);
}
grpc::ClientContext context_;
TestService::Stub *stub_;
TestService::Stub* stub_;
RequestType req_;
ResponseType response_;
bool (ClientRpcContextStreamingImpl::*next_state_)(bool, Histogram *);
std::function<void(grpc::Status, ResponseType *)> callback_;
std::function<std::unique_ptr<grpc::ClientAsyncReaderWriter<
RequestType,ResponseType>>(
TestService::Stub *, grpc::ClientContext *, void *)> start_req_;
bool (ClientRpcContextStreamingImpl::*next_state_)(bool, Histogram*);
std::function<void(grpc::Status, ResponseType*)> callback_;
std::function<
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
TestService::Stub*, grpc::ClientContext*, void*)> start_req_;
grpc::Status status_;
double start_;
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType,ResponseType>>
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>
stream_;
};
class AsyncStreamingClient GRPC_FINAL : public AsyncClient {
public:
explicit AsyncStreamingClient(const ClientConfig &config) :
AsyncClient(config, SetupCtx) {
explicit AsyncStreamingClient(const ClientConfig& config)
: AsyncClient(config, SetupCtx) {
StartThreads(config.async_client_threads());
}
~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
private:
private:
static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub,
const SimpleRequest& req) {
auto check_done = [](grpc::Status s, SimpleResponse* response) {};
auto start_req = [cq](TestService::Stub *stub, grpc::ClientContext *ctx,
void *tag) {
auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx,
void* tag) {
auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
return stream;
};

8
test/cpp/qps/client_sync.cc
Unescape View File

@ -99,7 +99,9 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
public:
SynchronousStreamingClient(const ClientConfig& config)
: SynchronousClient(config), context_(num_threads_), stream_(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]);
@ -128,8 +130,8 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
private:
std::vector<grpc::ClientContext> context_;
std::vector<std::unique_ptr<grpc::ClientReaderWriter<
SimpleRequest, SimpleResponse>>> stream_;
std::vector<std::unique_ptr<
grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>> stream_;
};
std::unique_ptr<Client> CreateSynchronousUnaryClient(

5
test/cpp/qps/driver.cc
Unescape View File

@ -105,7 +105,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
if (!called_init) {
char args_buf[100];
strcpy(args_buf, "some-benchmark");
char *args[] = {args_buf};
char* args[] = {args_buf};
grpc_test_init(1, args);
called_init = true;
}
@ -211,7 +211,8 @@ std::unique_ptr<ScenarioResult> RunScenario(
// Wait some time
gpr_log(GPR_INFO, "Running");
gpr_sleep_until(gpr_time_add(start, gpr_time_from_seconds(benchmark_seconds)));
gpr_sleep_until(
gpr_time_add(start, gpr_time_from_seconds(benchmark_seconds)));
// Finish a run
std::unique_ptr<ScenarioResult> result(new ScenarioResult);

4
test/cpp/qps/qps_driver.cc
Unescape View File

@ -100,8 +100,8 @@ int main(int argc, char** argv) {
// client will deadlock on a timer.
GPR_ASSERT(!(server_type == grpc::testing::SYNCHRONOUS_SERVER &&
rpc_type == grpc::testing::STREAMING &&
FLAGS_server_threads < FLAGS_client_channels *
FLAGS_outstanding_rpcs_per_channel));
FLAGS_server_threads <
FLAGS_client_channels * FLAGS_outstanding_rpcs_per_channel));
const auto result = RunScenario(
client_config, FLAGS_num_clients, server_config, FLAGS_num_servers,

24
test/cpp/qps/qps_worker.cc
Unescape View File

@ -64,17 +64,19 @@ namespace testing {
std::unique_ptr<Client> CreateClient(const ClientConfig& config) {
switch (config.client_type()) {
case ClientType::SYNCHRONOUS_CLIENT:
return (config.rpc_type() == RpcType::UNARY) ?
CreateSynchronousUnaryClient(config) :
CreateSynchronousStreamingClient(config);
return (config.rpc_type() == RpcType::UNARY)
? CreateSynchronousUnaryClient(config)
: CreateSynchronousStreamingClient(config);
case ClientType::ASYNC_CLIENT:
return (config.rpc_type() == RpcType::UNARY) ?
CreateAsyncUnaryClient(config) : CreateAsyncStreamingClient(config);
return (config.rpc_type() == RpcType::UNARY)
? CreateAsyncUnaryClient(config)
: CreateAsyncStreamingClient(config);
}
abort();
}
std::unique_ptr<Server> CreateServer(const ServerConfig& config, int server_port) {
std::unique_ptr<Server> CreateServer(const ServerConfig& config,
int server_port) {
switch (config.server_type()) {
case ServerType::SYNCHRONOUS_SERVER:
return CreateSynchronousServer(config, server_port);
@ -86,7 +88,8 @@ std::unique_ptr<Server> CreateServer(const ServerConfig& config, int server_port
class WorkerImpl GRPC_FINAL : public Worker::Service {
public:
explicit WorkerImpl(int server_port) : server_port_(server_port), acquired_(false) {}
explicit WorkerImpl(int server_port)
: server_port_(server_port), acquired_(false) {}
Status RunTest(ServerContext* ctx,
ServerReaderWriter<ClientStatus, ClientArgs>* stream)
@ -97,7 +100,7 @@ class WorkerImpl GRPC_FINAL : public Worker::Service {
}
grpc_profiler_start("qps_client.prof");
Status ret = RunTestBody(ctx,stream);
Status ret = RunTestBody(ctx, stream);
grpc_profiler_stop();
return ret;
}
@ -111,7 +114,7 @@ class WorkerImpl GRPC_FINAL : public Worker::Service {
}
grpc_profiler_start("qps_server.prof");
Status ret = RunServerBody(ctx,stream);
Status ret = RunServerBody(ctx, stream);
grpc_profiler_stop();
return ret;
}
@ -226,8 +229,7 @@ QpsWorker::QpsWorker(int driver_port, int server_port) {
server_ = std::move(builder.BuildAndStart());
}
QpsWorker::~QpsWorker() {
}
QpsWorker::~QpsWorker() {}
} // namespace testing
} // namespace grpc

12
test/cpp/qps/report.cc
Unescape View File

@ -48,18 +48,20 @@ void ReportQPS(const ScenarioResult& result) {
}
// QPS: XXX (YYY/server core)
void ReportQPSPerCore(const ScenarioResult& result, const ServerConfig& server_config) {
auto qps =
result.latencies.Count() /
void ReportQPSPerCore(const ScenarioResult& result,
const ServerConfig& server_config) {
auto qps = result.latencies.Count() /
average(result.client_resources,
[](ResourceUsage u) { return u.wall_time; });
gpr_log(GPR_INFO, "QPS: %.1f (%.1f/server core)", qps, qps/server_config.threads());
gpr_log(GPR_INFO, "QPS: %.1f (%.1f/server core)", qps,
qps / server_config.threads());
}
// Latency (50/90/95/99/99.9%-ile): AA/BB/CC/DD/EE us
void ReportLatency(const ScenarioResult& result) {
gpr_log(GPR_INFO, "Latencies (50/90/95/99/99.9%%-ile): %.1f/%.1f/%.1f/%.1f/%.1f us",
gpr_log(GPR_INFO,
"Latencies (50/90/95/99/99.9%%-ile): %.1f/%.1f/%.1f/%.1f/%.1f us",
result.latencies.Percentile(50) / 1000,
result.latencies.Percentile(90) / 1000,
result.latencies.Percentile(95) / 1000,

78
test/cpp/qps/server_async.cc
Unescape View File

@ -64,7 +64,7 @@ namespace testing {
class AsyncQpsServerTest : public Server {
public:
AsyncQpsServerTest(const ServerConfig &config, int port) : shutdown_(false) {
char* server_address = NULL;
char *server_address = NULL;
gpr_join_host_port(&server_address, "::", port);
ServerBuilder builder;
@ -95,9 +95,9 @@ class AsyncQpsServerTest : public Server {
threads_.push_back(std::thread([=]() {
// Wait until work is available or we are shutting down
bool ok;
void* got_tag;
void *got_tag;
while (srv_cq_->Next(&got_tag, &ok)) {
ServerRpcContext* ctx = detag(got_tag);
ServerRpcContext *ctx = detag(got_tag);
// The tag is a pointer to an RPC context to invoke
if (ctx->RunNextState(ok) == false) {
// this RPC context is done, so refresh it
@ -135,21 +135,21 @@ class AsyncQpsServerTest : public Server {
virtual bool RunNextState(bool) = 0; // next state, return false if done
virtual void Reset() = 0; // start this back at a clean state
};
static void* tag(ServerRpcContext* func) {
return reinterpret_cast<void*>(func);
static void *tag(ServerRpcContext *func) {
return reinterpret_cast<void *>(func);
}
static ServerRpcContext* detag(void* tag) {
return reinterpret_cast<ServerRpcContext*>(tag);
static ServerRpcContext *detag(void *tag) {
return reinterpret_cast<ServerRpcContext *>(tag);
}
template <class RequestType, class ResponseType>
class ServerRpcContextUnaryImpl GRPC_FINAL : public ServerRpcContext {
public:
ServerRpcContextUnaryImpl(
std::function<void(ServerContext*, RequestType*,
grpc::ServerAsyncResponseWriter<ResponseType>*,
void*)> request_method,
std::function<grpc::Status(const RequestType*, ResponseType*)>
std::function<void(ServerContext *, RequestType *,
grpc::ServerAsyncResponseWriter<ResponseType> *,
void *)> request_method,
std::function<grpc::Status(const RequestType *, ResponseType *)>
invoke_method)
: next_state_(&ServerRpcContextUnaryImpl::invoker),
request_method_(request_method),
@ -159,7 +159,9 @@ class AsyncQpsServerTest : public Server {
AsyncQpsServerTest::tag(this));
}
~ServerRpcContextUnaryImpl() GRPC_OVERRIDE {}
bool RunNextState(bool ok) GRPC_OVERRIDE {return (this->*next_state_)(ok);}
bool RunNextState(bool ok) GRPC_OVERRIDE {
return (this->*next_state_)(ok);
}
void Reset() GRPC_OVERRIDE {
srv_ctx_ = ServerContext();
req_ = RequestType();
@ -192,10 +194,10 @@ class AsyncQpsServerTest : public Server {
ServerContext srv_ctx_;
RequestType req_;
bool (ServerRpcContextUnaryImpl::*next_state_)(bool);
std::function<void(ServerContext*, RequestType*,
grpc::ServerAsyncResponseWriter<ResponseType>*, void*)>
std::function<void(ServerContext *, RequestType *,
grpc::ServerAsyncResponseWriter<ResponseType> *, void *)>
request_method_;
std::function<grpc::Status(const RequestType*, ResponseType*)>
std::function<grpc::Status(const RequestType *, ResponseType *)>
invoke_method_;
grpc::ServerAsyncResponseWriter<ResponseType> response_writer_;
};
@ -204,9 +206,9 @@ class AsyncQpsServerTest : public Server {
class ServerRpcContextStreamingImpl GRPC_FINAL : public ServerRpcContext {
public:
ServerRpcContextStreamingImpl(
std::function<void(ServerContext *,
grpc::ServerAsyncReaderWriter<ResponseType,
RequestType> *, void *)> request_method,
std::function<void(ServerContext *, grpc::ServerAsyncReaderWriter<
ResponseType, RequestType> *,
void *)> request_method,
std::function<grpc::Status(const RequestType *, ResponseType *)>
invoke_method)
: next_state_(&ServerRpcContextStreamingImpl::request_done),
@ -215,14 +217,15 @@ class AsyncQpsServerTest : public Server {
stream_(&srv_ctx_) {
request_method_(&srv_ctx_, &stream_, AsyncQpsServerTest::tag(this));
}
~ServerRpcContextStreamingImpl() GRPC_OVERRIDE {
~ServerRpcContextStreamingImpl() GRPC_OVERRIDE {}
bool RunNextState(bool ok) GRPC_OVERRIDE {
return (this->*next_state_)(ok);
}
bool RunNextState(bool ok) GRPC_OVERRIDE {return (this->*next_state_)(ok);}
void Reset() GRPC_OVERRIDE {
srv_ctx_ = ServerContext();
req_ = RequestType();
stream_ = grpc::ServerAsyncReaderWriter<ResponseType,
RequestType>(&srv_ctx_);
stream_ =
grpc::ServerAsyncReaderWriter<ResponseType, RequestType>(&srv_ctx_);
// Then request the method
next_state_ = &ServerRpcContextStreamingImpl::request_done;
@ -260,28 +263,28 @@ class AsyncQpsServerTest : public Server {
if (ok) {
stream_.Read(&req_, AsyncQpsServerTest::tag(this));
next_state_ = &ServerRpcContextStreamingImpl::read_done;
}
else {
} else {
stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
next_state_ = &ServerRpcContextStreamingImpl::finish_done;
}
return true;
}
bool finish_done(bool ok) {return false; /* reset the context */ }
bool finish_done(bool ok) { return false; /* reset the context */ }
ServerContext srv_ctx_;
RequestType req_;
bool (ServerRpcContextStreamingImpl::*next_state_)(bool);
std::function<void(ServerContext *,
grpc::ServerAsyncReaderWriter<ResponseType,
RequestType> *, void *)> request_method_;
std::function<void(
ServerContext *,
grpc::ServerAsyncReaderWriter<ResponseType, RequestType> *, void *)>
request_method_;
std::function<grpc::Status(const RequestType *, ResponseType *)>
invoke_method_;
grpc::ServerAsyncReaderWriter<ResponseType,RequestType> stream_;
grpc::ServerAsyncReaderWriter<ResponseType, RequestType> stream_;
};
static Status ProcessRPC(const SimpleRequest* request,
SimpleResponse* response) {
static Status ProcessRPC(const SimpleRequest *request,
SimpleResponse *response) {
if (request->response_size() > 0) {
if (!SetPayload(request->response_type(), request->response_size(),
response->mutable_payload())) {
@ -294,19 +297,20 @@ class AsyncQpsServerTest : public Server {
std::unique_ptr<grpc::Server> server_;
std::unique_ptr<grpc::ServerCompletionQueue> srv_cq_;
TestService::AsyncService async_service_;
std::function<void(ServerContext*, SimpleRequest*,
grpc::ServerAsyncResponseWriter<SimpleResponse>*, void*)>
std::function<void(ServerContext *, SimpleRequest *,
grpc::ServerAsyncResponseWriter<SimpleResponse> *, void *)>
request_unary_;
std::function<void(ServerContext*, grpc::ServerAsyncReaderWriter<
SimpleResponse,SimpleRequest>*, void*)>
std::function<void(
ServerContext *,
grpc::ServerAsyncReaderWriter<SimpleResponse, SimpleRequest> *, void *)>
request_streaming_;
std::forward_list<ServerRpcContext*> contexts_;
std::forward_list<ServerRpcContext *> contexts_;
std::mutex shutdown_mutex_;
bool shutdown_;
};
std::unique_ptr<Server> CreateAsyncServer(const ServerConfig& config,
std::unique_ptr<Server> CreateAsyncServer(const ServerConfig &config,
int port) {
return std::unique_ptr<Server>(new AsyncQpsServerTest(config, port));
}

6
test/cpp/qps/server_sync.cc
Unescape View File

@ -70,9 +70,9 @@ class TestServiceImpl GRPC_FINAL : public TestService::Service {
}
return Status::OK;
}
Status StreamingCall(ServerContext *context,
ServerReaderWriter<SimpleResponse, SimpleRequest>*
stream) GRPC_OVERRIDE {
Status StreamingCall(
ServerContext* context,
ServerReaderWriter<SimpleResponse, SimpleRequest>* stream) GRPC_OVERRIDE {
SimpleRequest request;
while (stream->Read(&request)) {
SimpleResponse response;

2
test/cpp/qps/worker.cc
Unescape View File

@ -48,7 +48,7 @@ DEFINE_int32(server_port, 0, "Spawned server port.");
static bool got_sigint = false;
static void sigint_handler(int x) {got_sigint = true;}
static void sigint_handler(int x) { got_sigint = true; }
namespace grpc {
namespace testing {

Write Preview
Loading…
Cancel
Save