aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp')
-rw-r--r--test/cpp/end2end/thread_stress_test.cc18
-rw-r--r--test/cpp/qps/client_async.cc34
-rw-r--r--test/cpp/qps/driver.cc13
-rw-r--r--test/cpp/qps/driver.h3
-rw-r--r--test/cpp/qps/qps_driver.cc7
-rw-r--r--test/cpp/qps/qps_worker.cc24
-rw-r--r--test/cpp/qps/qps_worker.h7
-rw-r--r--test/cpp/qps/worker.cc2
-rw-r--r--test/cpp/util/time_test.cc4
9 files changed, 78 insertions, 34 deletions
diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc
index 4c7caa9b87..e246c0b0e2 100644
--- a/test/cpp/end2end/thread_stress_test.cc
+++ b/test/cpp/end2end/thread_stress_test.cc
@@ -56,7 +56,9 @@ using grpc::testing::EchoResponse;
using std::chrono::system_clock;
const int kNumThreads = 100; // Number of threads
-const int kNumRpcs = 1000; // Number of RPCs per thread
+const int kNumAsyncSendThreads = 2;
+const int kNumAsyncReceiveThreads = 50;
+const int kNumRpcs = 1000; // Number of RPCs per thread
namespace grpc {
namespace testing {
@@ -273,7 +275,7 @@ class AsyncClientEnd2endTest : public ::testing::Test {
for (int i = 0; i < num_rpcs; ++i) {
AsyncClientCall* call = new AsyncClientCall;
EchoRequest request;
- request.set_message("Hello");
+ request.set_message("Hello: " + std::to_string(i));
call->response_reader =
common_.GetStub()->AsyncEcho(&call->context, request, &cq_);
call->response_reader->Finish(&call->response, &call->status,
@@ -290,7 +292,9 @@ class AsyncClientEnd2endTest : public ::testing::Test {
bool ok = false;
if (!cq_.Next(&got_tag, &ok)) break;
AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);
- GPR_ASSERT(ok);
+ if (!ok) {
+ gpr_log(GPR_DEBUG, "Error: %d", call->status.error_code());
+ }
delete call;
bool notify;
@@ -315,22 +319,22 @@ class AsyncClientEnd2endTest : public ::testing::Test {
TEST_F(AsyncClientEnd2endTest, ThreadStress) {
common_.ResetStub();
std::vector<std::thread*> send_threads, completion_threads;
- for (int i = 0; i < kNumThreads / 2; ++i) {
+ for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
completion_threads.push_back(new std::thread(
&AsyncClientEnd2endTest_ThreadStress_Test::AsyncCompleteRpc, this));
}
- for (int i = 0; i < kNumThreads / 2; ++i) {
+ for (int i = 0; i < kNumAsyncSendThreads; ++i) {
send_threads.push_back(
new std::thread(&AsyncClientEnd2endTest_ThreadStress_Test::AsyncSendRpc,
this, kNumRpcs));
}
- for (int i = 0; i < kNumThreads / 2; ++i) {
+ for (int i = 0; i < kNumAsyncSendThreads; ++i) {
send_threads[i]->join();
delete send_threads[i];
}
Wait();
- for (int i = 0; i < kNumThreads / 2; ++i) {
+ for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
completion_threads[i]->join();
delete completion_threads[i];
}
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 6b6294ba51..f3f8f37051 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -96,8 +96,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
std::function<
std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
- CompletionQueue*)>
- start_req,
+ CompletionQueue*)> start_req,
std::function<void(grpc::Status, ResponseType*)> on_done)
: ClientRpcContext(channel_id),
context_(),
@@ -143,8 +142,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
std::function<void(grpc::Status, ResponseType*)> callback_;
std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
- CompletionQueue*)>
- start_req_;
+ CompletionQueue*)> start_req_;
grpc::Status status_;
double start_;
std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>
@@ -164,12 +162,11 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
using ClientImpl<StubType, RequestType>::cores_;
using ClientImpl<StubType, RequestType>::channels_;
using ClientImpl<StubType, RequestType>::request_;
- AsyncClient(
- const ClientConfig& config,
- std::function<ClientRpcContext*(int, StubType*, const RequestType&)>
- setup_ctx,
- std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)>
- create_stub)
+ AsyncClient(const ClientConfig& config,
+ std::function<ClientRpcContext*(int, StubType*,
+ const RequestType&)> setup_ctx,
+ std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)>
+ create_stub)
: ClientImpl<StubType, RequestType>(config, create_stub),
num_async_threads_(NumThreads(config)),
channel_lock_(new std::mutex[config.client_channels()]),
@@ -411,8 +408,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
std::function<std::unique_ptr<
grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*,
- void*)>
- start_req,
+ void*)> start_req,
std::function<void(grpc::Status, ResponseType*)> on_done)
: ClientRpcContext(channel_id),
context_(),
@@ -464,10 +460,10 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext {
ResponseType response_;
bool (ClientRpcContextStreamingImpl::*next_state_)(bool, Histogram*);
std::function<void(grpc::Status, ResponseType*)> callback_;
- 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_;
grpc::Status status_;
double start_;
std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>
@@ -511,8 +507,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
int channel_id, grpc::GenericStub* stub, const ByteBuffer& req,
std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>(
grpc::GenericStub*, grpc::ClientContext*,
- const grpc::string& method_name, CompletionQueue*, void*)>
- start_req,
+ const grpc::string& method_name, CompletionQueue*, void*)> start_req,
std::function<void(grpc::Status, ByteBuffer*)> on_done)
: ClientRpcContext(channel_id),
context_(),
@@ -569,8 +564,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
std::function<void(grpc::Status, ByteBuffer*)> callback_;
std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>(
grpc::GenericStub*, grpc::ClientContext*, const grpc::string&,
- CompletionQueue*, void*)>
- start_req_;
+ CompletionQueue*, void*)> start_req_;
grpc::Status status_;
double start_;
std::unique_ptr<grpc::GenericClientAsyncReaderWriter> stream_;
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index c70b0303b8..80f6ada409 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -384,5 +384,18 @@ std::unique_ptr<ScenarioResult> RunScenario(
delete[] servers;
return result;
}
+
+void RunQuit() {
+ // Get client, server lists
+ auto workers = get_workers("QPS_WORKERS");
+ for (size_t i = 0; i < workers.size(); i++) {
+ auto stub = WorkerService::NewStub(
+ CreateChannel(workers[i], InsecureChannelCredentials()));
+ Void dummy;
+ grpc::ClientContext ctx;
+ GPR_ASSERT(stub->QuitWorker(&ctx, dummy, &dummy).ok());
+ }
+}
+
} // namespace testing
} // namespace grpc
diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h
index 2a7cf805e5..3af61f7391 100644
--- a/test/cpp/qps/driver.h
+++ b/test/cpp/qps/driver.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -70,6 +70,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
const grpc::testing::ServerConfig& server_config, size_t num_servers,
int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count);
+void RunQuit();
} // namespace testing
} // namespace grpc
diff --git a/test/cpp/qps/qps_driver.cc b/test/cpp/qps/qps_driver.cc
index ffc8a83fc5..69fb4d75e8 100644
--- a/test/cpp/qps/qps_driver.cc
+++ b/test/cpp/qps/qps_driver.cc
@@ -78,6 +78,8 @@ DEFINE_int32(client_core_limit, -1, "Limit on client cores to use");
DEFINE_bool(secure_test, false, "Run a secure test");
+DEFINE_bool(quit, false, "Quit the workers");
+
using grpc::testing::ClientConfig;
using grpc::testing::ServerConfig;
using grpc::testing::ClientType;
@@ -90,6 +92,11 @@ namespace grpc {
namespace testing {
static void QpsDriver() {
+ if (FLAGS_quit) {
+ RunQuit();
+ return;
+ }
+
RpcType rpc_type;
GPR_ASSERT(RpcType_Parse(FLAGS_rpc_type, &rpc_type));
diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc
index 7e9e05f7ec..9442017ddf 100644
--- a/test/cpp/qps/qps_worker.cc
+++ b/test/cpp/qps/qps_worker.cc
@@ -103,8 +103,8 @@ static std::unique_ptr<Server> CreateServer(const ServerConfig& config) {
class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
public:
- explicit WorkerServiceImpl(int server_port)
- : acquired_(false), server_port_(server_port) {}
+ WorkerServiceImpl(int server_port, QpsWorker* worker)
+ : acquired_(false), server_port_(server_port), worker_(worker) {}
Status RunClient(ServerContext* ctx,
ServerReaderWriter<ClientStatus, ClientArgs>* stream)
@@ -140,6 +140,16 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
return Status::OK;
}
+ Status QuitWorker(ServerContext* ctx, const Void*, Void*) GRPC_OVERRIDE {
+ InstanceGuard g(this);
+ if (!g.Acquired()) {
+ return Status(StatusCode::RESOURCE_EXHAUSTED, "");
+ }
+
+ worker_->MarkDone();
+ return Status::OK;
+ }
+
private:
// Protect against multiple clients using this worker at once.
class InstanceGuard {
@@ -250,10 +260,12 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service {
std::mutex mu_;
bool acquired_;
int server_port_;
+ QpsWorker* worker_;
};
QpsWorker::QpsWorker(int driver_port, int server_port) {
- impl_.reset(new WorkerServiceImpl(server_port));
+ impl_.reset(new WorkerServiceImpl(server_port, this));
+ gpr_atm_rel_store(&done_, static_cast<gpr_atm>(0));
char* server_address = NULL;
gpr_join_host_port(&server_address, "::", driver_port);
@@ -269,5 +281,11 @@ QpsWorker::QpsWorker(int driver_port, int server_port) {
QpsWorker::~QpsWorker() {}
+bool QpsWorker::Done() const {
+ return (gpr_atm_acq_load(&done_) != static_cast<gpr_atm>(0));
+}
+void QpsWorker::MarkDone() {
+ gpr_atm_rel_store(&done_, static_cast<gpr_atm>(1));
+}
} // namespace testing
} // namespace grpc
diff --git a/test/cpp/qps/qps_worker.h b/test/cpp/qps/qps_worker.h
index 27de69fa65..624c182100 100644
--- a/test/cpp/qps/qps_worker.h
+++ b/test/cpp/qps/qps_worker.h
@@ -36,6 +36,8 @@
#include <memory>
+#include <grpc/support/atm.h>
+
namespace grpc {
class Server;
@@ -49,9 +51,14 @@ class QpsWorker {
explicit QpsWorker(int driver_port, int server_port = 0);
~QpsWorker();
+ bool Done() const;
+ void MarkDone();
+
private:
std::unique_ptr<WorkerServiceImpl> impl_;
std::unique_ptr<Server> server_;
+
+ gpr_atm done_;
};
} // namespace testing
diff --git a/test/cpp/qps/worker.cc b/test/cpp/qps/worker.cc
index a1e73e9abe..f42cfe3255 100644
--- a/test/cpp/qps/worker.cc
+++ b/test/cpp/qps/worker.cc
@@ -56,7 +56,7 @@ namespace testing {
static void RunServer() {
QpsWorker worker(FLAGS_driver_port, FLAGS_server_port);
- while (!got_sigint) {
+ while (!got_sigint && !worker.Done()) {
gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_seconds(5, GPR_TIMESPAN)));
}
diff --git a/test/cpp/util/time_test.cc b/test/cpp/util/time_test.cc
index 1e501dfd28..48c6ce7697 100644
--- a/test/cpp/util/time_test.cc
+++ b/test/cpp/util/time_test.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -45,7 +45,7 @@ namespace {
class TimeTest : public ::testing::Test {};
TEST_F(TimeTest, AbsolutePointTest) {
- long us = 10000000L;
+ int64_t us = 10000000L;
gpr_timespec ts = gpr_time_from_micros(us, GPR_TIMESPAN);
ts.clock_type = GPR_CLOCK_REALTIME;
system_clock::time_point tp{microseconds(us)};
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-21 14:39:09 +0000