aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp/qps
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp/qps')
-rw-r--r--test/cpp/qps/client.h43
-rw-r--r--test/cpp/qps/client_async.cc89
-rw-r--r--test/cpp/qps/client_sync.cc14
-rw-r--r--test/cpp/qps/driver.cc5
-rw-r--r--test/cpp/qps/qps_driver.cc4
-rw-r--r--test/cpp/qps/qps_worker.cc24
-rw-r--r--test/cpp/qps/server_async.cc110
-rw-r--r--test/cpp/qps/server_sync.cc16
-rw-r--r--test/cpp/qps/worker.cc7
9 files changed, 163 insertions, 149 deletions
diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h
index 2dc5b3860f..dc3a9f2ac5 100644
--- a/test/cpp/qps/client.h
+++ b/test/cpp/qps/client.h
@@ -113,24 +113,26 @@ class Client {
: done_(false),
new_(nullptr),
impl_([this, idx, client]() {
- for (;;) {
- // run the loop body
- bool thread_still_ok = client->ThreadFunc(&histogram_, idx);
- // lock, see if we're done
- std::lock_guard<std::mutex> g(mu_);
- if (!thread_still_ok) {
- gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
- done_ = true;
- }
- if (done_) {return;}
- // check if we're marking, swap out the histogram if so
- if (new_) {
- new_->Swap(&histogram_);
- new_ = nullptr;
- cv_.notify_one();
- }
+ for (;;) {
+ // run the loop body
+ bool thread_still_ok = client->ThreadFunc(&histogram_, idx);
+ // lock, see if we're done
+ std::lock_guard<std::mutex> g(mu_);
+ if (!thread_still_ok) {
+ gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
+ done_ = true;
}
- }) {}
+ if (done_) {
+ return;
+ }
+ // check if we're marking, swap out the histogram if so
+ if (new_) {
+ new_->Swap(&histogram_);
+ new_ = nullptr;
+ cv_.notify_one();
+ }
+ }
+ }) {}
~Thread() {
{
@@ -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);
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index e3ab57728d..00bbd8a8a0 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -128,16 +128,16 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
class AsyncClient : public Client {
public:
explicit AsyncClient(const ClientConfig& config,
- std::function<void(CompletionQueue*, TestService::Stub*,
- const SimpleRequest&)> setup_ctx) :
- Client(config) {
+ std::function<void(CompletionQueue*, TestService::Stub*,
+ const SimpleRequest&)> setup_ctx)
+ : Client(config) {
for (int i = 0; i < config.async_client_threads(); i++) {
cli_cqs_.emplace_back(new CompletionQueue);
}
int t = 0;
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
for (auto channel = channels_.begin(); channel != channels_.end();
- channel++) {
+ channel++) {
auto* cq = cli_cqs_[t].get();
t = (t + 1) % cli_cqs_.size();
setup_ctx(cq, channel->get_stub(), request_);
@@ -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>>
- stream_;
+ 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) {
+ 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;
};
diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc
index 264293561d..c28dc91321 100644
--- a/test/cpp/qps/client_sync.cc
+++ b/test/cpp/qps/client_sync.cc
@@ -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]);
@@ -110,8 +112,8 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
EndThreads();
for (auto stream = stream_.begin(); stream != stream_.end(); stream++) {
if (*stream) {
- (*stream)->WritesDone();
- EXPECT_TRUE((*stream)->Finish().IsOk());
+ (*stream)->WritesDone();
+ EXPECT_TRUE((*stream)->Finish().IsOk());
}
}
}
@@ -119,7 +121,7 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
double start = Timer::Now();
if (stream_[thread_idx]->Write(request_) &&
- stream_[thread_idx]->Read(&responses_[thread_idx])) {
+ stream_[thread_idx]->Read(&responses_[thread_idx])) {
histogram->Add((Timer::Now() - start) * 1e9);
return true;
}
@@ -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(
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index 682d515ebe..bf12730f97 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -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);
diff --git a/test/cpp/qps/qps_driver.cc b/test/cpp/qps/qps_driver.cc
index 0b6759a6ed..281e2e8119 100644
--- a/test/cpp/qps/qps_driver.cc
+++ b/test/cpp/qps/qps_driver.cc
@@ -104,8 +104,8 @@ static void QpsDriver() {
// 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,
diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc
index 46d70dce52..fb49271991 100644
--- a/test/cpp/qps/qps_worker.cc
+++ b/test/cpp/qps/qps_worker.cc
@@ -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
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index 6cb3192908..58499f345f 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -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
@@ -133,23 +133,23 @@ class AsyncQpsServerTest : public Server {
ServerRpcContext() {}
virtual ~ServerRpcContext(){};
virtual bool RunNextState(bool) = 0; // next state, return false if done
- virtual void Reset() = 0; // start this back at a clean state
+ 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;
@@ -241,47 +244,47 @@ class AsyncQpsServerTest : public Server {
bool read_done(bool ok) {
if (ok) {
- // invoke the method
- ResponseType response;
- // Call the RPC processing function
- grpc::Status status = invoke_method_(&req_, &response);
- // initiate the write
- stream_.Write(response, AsyncQpsServerTest::tag(this));
- next_state_ = &ServerRpcContextStreamingImpl::write_done;
- } else { // client has sent writes done
- // finish the stream
- stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
- next_state_ = &ServerRpcContextStreamingImpl::finish_done;
+ // invoke the method
+ ResponseType response;
+ // Call the RPC processing function
+ grpc::Status status = invoke_method_(&req_, &response);
+ // initiate the write
+ stream_.Write(response, AsyncQpsServerTest::tag(this));
+ next_state_ = &ServerRpcContextStreamingImpl::write_done;
+ } else { // client has sent writes done
+ // finish the stream
+ stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
+ next_state_ = &ServerRpcContextStreamingImpl::finish_done;
}
return true;
}
bool write_done(bool ok) {
// now go back and get another streaming read!
if (ok) {
- stream_.Read(&req_, AsyncQpsServerTest::tag(this));
- next_state_ = &ServerRpcContextStreamingImpl::read_done;
- }
- else {
- stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
- next_state_ = &ServerRpcContextStreamingImpl::finish_done;
+ stream_.Read(&req_, AsyncQpsServerTest::tag(this));
+ next_state_ = &ServerRpcContextStreamingImpl::read_done;
+ } 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));
}
diff --git a/test/cpp/qps/server_sync.cc b/test/cpp/qps/server_sync.cc
index 2770233a7c..bc00de9ced 100644
--- a/test/cpp/qps/server_sync.cc
+++ b/test/cpp/qps/server_sync.cc
@@ -70,18 +70,18 @@ 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;
if (request.response_size() > 0) {
- if (!Server::SetPayload(request.response_type(),
- request.response_size(),
- response.mutable_payload())) {
- return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
- }
+ if (!Server::SetPayload(request.response_type(),
+ request.response_size(),
+ response.mutable_payload())) {
+ return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+ }
}
stream->Write(response);
}
diff --git a/test/cpp/qps/worker.cc b/test/cpp/qps/worker.cc
index 281c617382..dfc102fc17 100644
--- a/test/cpp/qps/worker.cc
+++ b/test/cpp/qps/worker.cc
@@ -37,6 +37,7 @@
#include <thread>
#include <grpc/grpc.h>
+#include <grpc/support/time.h>
#include <gflags/gflags.h>
#include "qps_worker.h"
@@ -47,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 {
@@ -56,7 +57,7 @@ static void RunServer() {
QpsWorker worker(FLAGS_driver_port, FLAGS_server_port);
while (!got_sigint) {
- std::this_thread::sleep_for(std::chrono::seconds(5));
+ gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(5)));
}
}
@@ -69,6 +70,6 @@ int main(int argc, char** argv) {
signal(SIGINT, sigint_handler);
grpc::testing::RunServer();
-
+
return 0;
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-21 09:50:41 +0000