diff --git a/Makefile b/Makefile
index 36eedb95added1f2a72ac4938f9b28dc15678ddf..aef6ff3a63933d37211a9df16cca22fd0a289561 100644
--- a/Makefile
+++ b/Makefile
@@ -8181,6 +8181,7 @@ QPS_WORKER_SRC = \
test/cpp/qps/client.cc \
test/cpp/qps/client_async.cc \
test/cpp/qps/server.cc \
+ test/cpp/qps/server_async.cc \
test/cpp/qps/worker.cc \
QPS_WORKER_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(QPS_WORKER_SRC))))
@@ -8214,6 +8215,7 @@ endif
$(OBJDIR)/$(CONFIG)/test/cpp/qps/client.o: $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(OBJDIR)/$(CONFIG)/test/cpp/qps/client_async.o: $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(OBJDIR)/$(CONFIG)/test/cpp/qps/server.o: $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
+$(OBJDIR)/$(CONFIG)/test/cpp/qps/server_async.o: $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(OBJDIR)/$(CONFIG)/test/cpp/qps/worker.o: $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
deps_qps_worker: $(QPS_WORKER_OBJS:.o=.dep)
diff --git a/build.json b/build.json
index 08b1e1842c3bcc6ab941ddf645d0b966c8ecd3c7..a8dc2e6bec547a342a97faa9902aa19ebf7b0e0e 100644
--- a/build.json
+++ b/build.json
@@ -1849,6 +1849,7 @@
"test/cpp/qps/client.cc",
"test/cpp/qps/client_async.cc",
"test/cpp/qps/server.cc",
+ "test/cpp/qps/server_async.cc",
"test/cpp/qps/worker.cc"
],
"deps": [
diff --git a/test/cpp/qps/server.cc b/test/cpp/qps/server.cc
index 8424dbafe975e5e58e7619f79c8af3152fae1085..e598fb51ae030dcb238881179720027ab2320ea0 100644
--- a/test/cpp/qps/server.cc
+++ b/test/cpp/qps/server.cc
@@ -57,24 +57,12 @@
namespace grpc {
namespace testing {
-static bool SetPayload(PayloadType type, int size, Payload* payload) {
- PayloadType response_type = type;
- // TODO(yangg): Support UNCOMPRESSABLE payload.
- if (type != PayloadType::COMPRESSABLE) {
- return false;
- }
- payload->set_type(response_type);
- std::unique_ptr<char[]> body(new char[size]());
- payload->set_body(body.get(), size);
- return true;
-}
-
class TestServiceImpl GRPC_FINAL : public TestService::Service {
public:
Status UnaryCall(ServerContext* context, const SimpleRequest* request,
SimpleResponse* response) override {
if (request->has_response_size() && request->response_size() > 0) {
- if (!SetPayload(request->response_type(), request->response_size(),
+ if (!Server::SetPayload(request->response_type(), request->response_size(),
response->mutable_payload())) {
return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
}
@@ -87,21 +75,7 @@ class SynchronousServer GRPC_FINAL : public grpc::testing::Server {
public:
SynchronousServer(const ServerConfig& config, int port)
: thread_pool_(config.threads()),
- impl_(MakeImpl(port)),
- timer_(new Timer) {}
-
- ServerStats Mark() GRPC_OVERRIDE {
- std::unique_ptr<Timer> timer(new Timer);
- timer.swap(timer_);
-
- auto timer_result = timer->Mark();
-
- ServerStats stats;
- stats.set_time_elapsed(timer_result.wall);
- stats.set_time_system(timer_result.system);
- stats.set_time_user(timer_result.user);
- return stats;
- }
+ impl_(MakeImpl(port)) {}
private:
std::unique_ptr<grpc::Server> MakeImpl(int port) {
@@ -120,7 +94,6 @@ class SynchronousServer GRPC_FINAL : public grpc::testing::Server {
TestServiceImpl service_;
ThreadPool thread_pool_;
std::unique_ptr<grpc::Server> impl_;
- std::unique_ptr<Timer> timer_;
};
std::unique_ptr<grpc::testing::Server> CreateSynchronousServer(
diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h
index 3542c17a6a2dde6a1d9cc4e742e0219ab07db011..ca22d7ca1cda9cbe7b70b32a80541851e44c920c 100644
--- a/test/cpp/qps/server.h
+++ b/test/cpp/qps/server.h
@@ -34,6 +34,7 @@
#ifndef TEST_QPS_SERVER_H
#define TEST_QPS_SERVER_H
+#include "test/cpp/qps/timer.h"
#include "test/cpp/qps/qpstest.pb.h"
namespace grpc {
@@ -41,9 +42,36 @@ namespace testing {
class Server {
public:
+ Server():timer_(new Timer) {}
virtual ~Server() {}
- virtual ServerStats Mark() = 0;
+ ServerStats Mark() {
+ std::unique_ptr<Timer> timer(new Timer);
+ timer.swap(timer_);
+
+ auto timer_result = timer->Mark();
+
+ ServerStats stats;
+ stats.set_time_elapsed(timer_result.wall);
+ stats.set_time_system(timer_result.system);
+ stats.set_time_user(timer_result.user);
+ return stats;
+ }
+
+ static bool SetPayload(PayloadType type, int size, Payload* payload) {
+ PayloadType response_type = type;
+ // TODO(yangg): Support UNCOMPRESSABLE payload.
+ if (type != PayloadType::COMPRESSABLE) {
+ return false;
+ }
+ payload->set_type(response_type);
+ std::unique_ptr<char[]> body(new char[size]());
+ payload->set_body(body.get(), size);
+ return true;
+ }
+
+ private:
+ std::unique_ptr<Timer> timer_;
};
std::unique_ptr<Server> CreateSynchronousServer(const ServerConfig& config,
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index c006262fc34eb26fd001f128deaf31d03a31c6e1..741a85802adb328566f852346ad7ec314d18e253 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -51,104 +51,37 @@
#include "src/cpp/server/thread_pool.h"
#include "test/core/util/grpc_profiler.h"
#include "test/cpp/qps/qpstest.pb.h"
+#include "test/cpp/qps/server.h"
#include <grpc/grpc.h>
#include <grpc/support/log.h>
-DEFINE_bool(enable_ssl, false, "Whether to use ssl/tls.");
-DEFINE_int32(port, 0, "Server port.");
-DEFINE_int32(server_threads, 4, "Number of server threads.");
+namespace grpc {
+ namespace testing {
-using grpc::CompletionQueue;
-using grpc::Server;
-using grpc::ServerBuilder;
-using grpc::ServerContext;
-using grpc::ThreadPool;
-using grpc::testing::Payload;
-using grpc::testing::PayloadType;
-using grpc::testing::ServerStats;
-using grpc::testing::SimpleRequest;
-using grpc::testing::SimpleResponse;
-using grpc::testing::StatsRequest;
-using grpc::testing::TestService;
-using grpc::Status;
-
-// In some distros, gflags is in the namespace google, and in some others,
-// in gflags. This hack is enabling us to find both.
-namespace google {}
-namespace gflags {}
-using namespace google;
-using namespace gflags;
-
-static bool got_sigint = false;
-
-static void sigint_handler(int x) { got_sigint = 1; }
-
-static double time_double(struct timeval *tv) {
- return tv->tv_sec + 1e-6 * tv->tv_usec;
-}
-
-static bool SetPayload(PayloadType type, int size, Payload *payload) {
- PayloadType response_type = type;
- // TODO(yangg): Support UNCOMPRESSABLE payload.
- if (type != PayloadType::COMPRESSABLE) {
- return false;
- }
- payload->set_type(response_type);
- std::unique_ptr<char[]> body(new char[size]());
- payload->set_body(body.get(), size);
- return true;
-}
-
-namespace {
-
-class AsyncQpsServerTest {
+class AsyncQpsServerTest : public Server {
public:
- AsyncQpsServerTest() : srv_cq_(), async_service_(&srv_cq_), server_(nullptr) {
+ AsyncQpsServerTest(const ServerConfig& config, int port) : srv_cq_(), async_service_(&srv_cq_), server_(nullptr) {
char *server_address = NULL;
- gpr_join_host_port(&server_address, "::", FLAGS_port);
+ gpr_join_host_port(&server_address, "::", port);
ServerBuilder builder;
builder.AddPort(server_address);
+ gpr_free(server_address);
builder.RegisterAsyncService(&async_service_);
server_ = builder.BuildAndStart();
- gpr_log(GPR_INFO, "Server listening on %s\n", server_address);
- gpr_free(server_address);
using namespace std::placeholders;
request_unary_ = std::bind(&TestService::AsyncService::RequestUnaryCall,
&async_service_, _1, _2, _3, &srv_cq_, _4);
- request_stats_ =
- std::bind(&TestService::AsyncService::RequestCollectServerStats,
- &async_service_, _1, _2, _3, &srv_cq_, _4);
for (int i = 0; i < 100; i++) {
contexts_.push_front(
new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
request_unary_, UnaryCall));
- contexts_.push_front(
- new ServerRpcContextUnaryImpl<StatsRequest, ServerStats>(
- request_stats_, CollectServerStats));
- }
- }
- ~AsyncQpsServerTest() {
- server_->Shutdown();
- void *ignored_tag;
- bool ignored_ok;
- srv_cq_.Shutdown();
- while (srv_cq_.Next(&ignored_tag, &ignored_ok)) {
}
- while (!contexts_.empty()) {
- delete contexts_.front();
- contexts_.pop_front();
- }
- for (auto& thr: threads_) {
- thr.join();
- }
- }
- void ServeRpcs(int num_threads) {
- for (int i = 0; i < num_threads; i++) {
+ for (int i = 0; i < config.threads(); i++) {
threads_.push_back(std::thread([=]() {
// Wait until work is available or we are shutting down
bool ok;
@@ -166,8 +99,16 @@ class AsyncQpsServerTest {
return;
}));
}
- while (!got_sigint) {
- std::this_thread::sleep_for(std::chrono::seconds(5));
+ }
+ ~AsyncQpsServerTest() {
+ server_->Shutdown();
+ srv_cq_.Shutdown();
+ for (auto& thr: threads_) {
+ thr.join();
+ }
+ while (!contexts_.empty()) {
+ delete contexts_.front();
+ contexts_.pop_front();
}
}
@@ -240,17 +181,6 @@ class AsyncQpsServerTest {
grpc::ServerAsyncResponseWriter<ResponseType> response_writer_;
};
- static Status CollectServerStats(const StatsRequest *,
- ServerStats *response) {
- struct rusage usage;
- struct timeval tv;
- gettimeofday(&tv, NULL);
- getrusage(RUSAGE_SELF, &usage);
- response->set_time_now(time_double(&tv));
- response->set_time_user(time_double(&usage.ru_utime));
- response->set_time_system(time_double(&usage.ru_stime));
- return Status::OK;
- }
static Status UnaryCall(const SimpleRequest *request,
SimpleResponse *response) {
if (request->has_response_size() && request->response_size() > 0) {
@@ -264,40 +194,16 @@ class AsyncQpsServerTest {
CompletionQueue srv_cq_;
TestService::AsyncService async_service_;
std::vector<std::thread> threads_;
- std::unique_ptr<Server> server_;
+ std::unique_ptr<grpc::Server> server_;
std::function<void(ServerContext *, SimpleRequest *,
grpc::ServerAsyncResponseWriter<SimpleResponse> *, void *)>
request_unary_;
- std::function<void(ServerContext *, StatsRequest *,
- grpc::ServerAsyncResponseWriter<ServerStats> *, void *)>
- request_stats_;
std::forward_list<ServerRpcContext *> contexts_;
};
-} // namespace
-
-static void RunServer() {
- AsyncQpsServerTest server;
-
- grpc_profiler_start("qps_server_async.prof");
-
- server.ServeRpcs(FLAGS_server_threads);
-
- grpc_profiler_stop();
+std::unique_ptr<Server> CreateAsyncServer(const ServerConfig& config, int port) {
+ return std::unique_ptr<Server>(new AsyncQpsServerTest(config, port));
}
-int main(int argc, char **argv) {
- grpc_init();
- ParseCommandLineFlags(&argc, &argv, true);
- GPR_ASSERT(FLAGS_port != 0);
- GPR_ASSERT(!FLAGS_enable_ssl);
-
- signal(SIGINT, sigint_handler);
-
- RunServer();
-
- grpc_shutdown();
- google::protobuf::ShutdownProtobufLibrary();
-
- return 0;
-}
+ }// namespace testing
+}// namespace grpc
diff --git a/test/cpp/qps/worker.cc b/test/cpp/qps/worker.cc
index 4a2e798a477b8d091faca6f1a8dfff574b179bcd..a8d5752120d3a8ff22a7238be064ad0cd11a8ea3 100644
--- a/test/cpp/qps/worker.cc
+++ b/test/cpp/qps/worker.cc
@@ -90,7 +90,7 @@ std::unique_ptr<Server> CreateServer(const ServerConfig& config) {
case ServerType::SYNCHRONOUS_SERVER:
return CreateSynchronousServer(config, FLAGS_server_port);
case ServerType::ASYNC_SERVER:
- abort(); // return CreateAsyncServer(config, FLAGS_server_port);
+ return CreateAsyncServer(config, FLAGS_server_port);
}
abort();
}