aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp')
-rw-r--r--test/cpp/common/channel_arguments_test.cc65
-rw-r--r--test/cpp/end2end/async_end2end_test.cc18
-rw-r--r--test/cpp/end2end/end2end_test.cc21
-rw-r--r--test/cpp/end2end/round_robin_end2end_test.cc9
-rw-r--r--test/cpp/end2end/test_service_impl.cc12
-rw-r--r--test/cpp/end2end/thread_stress_test.cc39
-rw-r--r--test/cpp/grpclb/grpclb_test.cc104
-rw-r--r--test/cpp/microbenchmarks/bm_fullstack.cc450
-rw-r--r--test/cpp/qps/client.h7
-rw-r--r--test/cpp/qps/client_async.cc1
-rw-r--r--test/cpp/qps/client_sync.cc14
-rw-r--r--test/cpp/qps/driver.cc72
-rwxr-xr-xtest/cpp/qps/gen_build_yaml.py61
-rw-r--r--test/cpp/qps/json_run_localhost.cc73
-rw-r--r--test/cpp/qps/qps_json_driver.cc142
-rw-r--r--test/cpp/qps/report.cc15
-rw-r--r--test/cpp/qps/report.h6
-rw-r--r--test/cpp/qps/server.h2
-rw-r--r--test/cpp/qps/usage_timer.cc33
-rw-r--r--test/cpp/qps/usage_timer.h2
-rw-r--r--test/cpp/util/config_grpc_cli.h2
-rw-r--r--test/cpp/util/grpc_tool_test.cc4
22 files changed, 947 insertions, 205 deletions
diff --git a/test/cpp/common/channel_arguments_test.cc b/test/cpp/common/channel_arguments_test.cc
index 342f6a5339..60d3215265 100644
--- a/test/cpp/common/channel_arguments_test.cc
+++ b/test/cpp/common/channel_arguments_test.cc
@@ -35,11 +35,56 @@
#include <grpc++/grpc++.h>
#include <grpc/grpc.h>
+#include <grpc/support/useful.h>
#include <gtest/gtest.h>
+#include "src/core/lib/iomgr/socket_mutator.h"
namespace grpc {
namespace testing {
+namespace {
+
+// A simple grpc_socket_mutator to be used to test SetSocketMutator
+class TestSocketMutator : public grpc_socket_mutator {
+ public:
+ TestSocketMutator();
+
+ bool MutateFd(int fd) {
+ // Do nothing on the fd
+ return true;
+ }
+};
+
+//
+// C API for TestSocketMutator
+//
+
+bool test_mutator_mutate_fd(int fd, grpc_socket_mutator* mutator) {
+ TestSocketMutator* tsm = (TestSocketMutator*)mutator;
+ return tsm->MutateFd(fd);
+}
+
+int test_mutator_compare(grpc_socket_mutator* a, grpc_socket_mutator* b) {
+ return GPR_ICMP(a, b);
+}
+
+void test_mutator_destroy(grpc_socket_mutator* mutator) {
+ TestSocketMutator* tsm = (TestSocketMutator*)mutator;
+ delete tsm;
+}
+
+grpc_socket_mutator_vtable test_mutator_vtable = {
+ test_mutator_mutate_fd, test_mutator_compare, test_mutator_destroy};
+
+//
+// TestSocketMutator implementation
+//
+
+TestSocketMutator::TestSocketMutator() {
+ grpc_socket_mutator_init(this, &test_mutator_vtable);
+}
+}
+
class ChannelArgumentsTest : public ::testing::Test {
protected:
ChannelArgumentsTest()
@@ -166,6 +211,26 @@ TEST_F(ChannelArgumentsTest, SetPointer) {
EXPECT_TRUE(HasArg(arg0));
}
+TEST_F(ChannelArgumentsTest, SetSocketMutator) {
+ VerifyDefaultChannelArgs();
+ grpc_arg arg0, arg1;
+ TestSocketMutator* mutator0 = new TestSocketMutator();
+ TestSocketMutator* mutator1 = new TestSocketMutator();
+ arg0 = grpc_socket_mutator_to_arg(mutator0);
+ arg1 = grpc_socket_mutator_to_arg(mutator1);
+
+ channel_args_.SetSocketMutator(mutator0);
+ EXPECT_TRUE(HasArg(arg0));
+
+ channel_args_.SetSocketMutator(mutator1);
+ EXPECT_TRUE(HasArg(arg1));
+ // arg0 is replaced by arg1
+ EXPECT_FALSE(HasArg(arg0));
+
+ // arg0 is destroyed by grpc_socket_mutator_to_arg(mutator1)
+ arg1.value.pointer.vtable->destroy(arg1.value.pointer.p);
+}
+
TEST_F(ChannelArgumentsTest, SetUserAgentPrefix) {
VerifyDefaultChannelArgs();
grpc::string prefix("prefix");
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index 3845582d5d..8e385d100c 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -352,15 +352,13 @@ void ServerWait(Server* server, int* notify) {
}
TEST_P(AsyncEnd2endTest, WaitAndShutdownTest) {
int notify = 0;
- std::thread* wait_thread =
- new std::thread(&ServerWait, server_.get(), &notify);
+ std::thread wait_thread(&ServerWait, server_.get(), &notify);
ResetStub();
SendRpc(1);
EXPECT_EQ(0, notify);
server_->Shutdown();
- wait_thread->join();
+ wait_thread.join();
EXPECT_EQ(1, notify);
- delete wait_thread;
}
TEST_P(AsyncEnd2endTest, ShutdownThenWait) {
@@ -991,7 +989,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
expected_server_cq_result = false;
}
- std::thread* server_try_cancel_thd = NULL;
+ std::thread* server_try_cancel_thd = nullptr;
auto verif = Verifier(GetParam().disable_blocking);
@@ -1027,7 +1025,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
}
}
- if (server_try_cancel_thd != NULL) {
+ if (server_try_cancel_thd != nullptr) {
server_try_cancel_thd->join();
delete server_try_cancel_thd;
}
@@ -1112,7 +1110,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
expected_cq_result = false;
}
- std::thread* server_try_cancel_thd = NULL;
+ std::thread* server_try_cancel_thd = nullptr;
auto verif = Verifier(GetParam().disable_blocking);
@@ -1150,7 +1148,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
}
}
- if (server_try_cancel_thd != NULL) {
+ if (server_try_cancel_thd != nullptr) {
server_try_cancel_thd->join();
delete server_try_cancel_thd;
}
@@ -1252,7 +1250,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
expected_cq_result = false;
}
- std::thread* server_try_cancel_thd = NULL;
+ std::thread* server_try_cancel_thd = nullptr;
auto verif = Verifier(GetParam().disable_blocking);
@@ -1332,7 +1330,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
EXPECT_EQ(verif.Next(cq_.get(), ignore_cq_result), 8);
}
- if (server_try_cancel_thd != NULL) {
+ if (server_try_cancel_thd != nullptr) {
server_try_cancel_thd->join();
delete server_try_cancel_thd;
}
diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 4b8749884f..9bb892c694 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -656,25 +656,23 @@ TEST_P(End2endTest, SimpleRpcWithCustomeUserAgentPrefix) {
TEST_P(End2endTest, MultipleRpcsWithVariedBinaryMetadataValue) {
ResetStub();
- std::vector<std::thread*> threads;
+ std::vector<std::thread> threads;
for (int i = 0; i < 10; ++i) {
- threads.push_back(new std::thread(SendRpc, stub_.get(), 10, true));
+ threads.emplace_back(SendRpc, stub_.get(), 10, true);
}
for (int i = 0; i < 10; ++i) {
- threads[i]->join();
- delete threads[i];
+ threads[i].join();
}
}
TEST_P(End2endTest, MultipleRpcs) {
ResetStub();
- std::vector<std::thread*> threads;
+ std::vector<std::thread> threads;
for (int i = 0; i < 10; ++i) {
- threads.push_back(new std::thread(SendRpc, stub_.get(), 10, false));
+ threads.emplace_back(SendRpc, stub_.get(), 10, false);
}
for (int i = 0; i < 10; ++i) {
- threads[i]->join();
- delete threads[i];
+ threads[i].join();
}
}
@@ -1058,13 +1056,12 @@ TEST_P(ProxyEnd2endTest, SimpleRpcWithEmptyMessages) {
TEST_P(ProxyEnd2endTest, MultipleRpcs) {
ResetStub();
- std::vector<std::thread*> threads;
+ std::vector<std::thread> threads;
for (int i = 0; i < 10; ++i) {
- threads.push_back(new std::thread(SendRpc, stub_.get(), 10, false));
+ threads.emplace_back(SendRpc, stub_.get(), 10, false);
}
for (int i = 0; i < 10; ++i) {
- threads[i]->join();
- delete threads[i];
+ threads[i].join();
}
}
diff --git a/test/cpp/end2end/round_robin_end2end_test.cc b/test/cpp/end2end/round_robin_end2end_test.cc
index 76211cbdd3..cc340b96b3 100644
--- a/test/cpp/end2end/round_robin_end2end_test.cc
+++ b/test/cpp/end2end/round_robin_end2end_test.cc
@@ -109,9 +109,9 @@ class RoundRobinEnd2endTest : public ::testing::Test {
uri << "127.0.0.1:" << servers_[i]->port_ << ",";
}
uri << "127.0.0.1:" << servers_[servers_.size() - 1]->port_;
- std::shared_ptr<Channel> channel =
+ channel_ =
CreateCustomChannel(uri.str(), InsecureChannelCredentials(), args);
- stub_ = grpc::testing::EchoTestService::NewStub(channel);
+ stub_ = grpc::testing::EchoTestService::NewStub(channel_);
}
void SendRpc(int num_rpcs) {
@@ -165,6 +165,7 @@ class RoundRobinEnd2endTest : public ::testing::Test {
const grpc::string server_host_;
CompletionQueue cli_cq_;
+ std::shared_ptr<Channel> channel_;
std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
std::vector<std::unique_ptr<ServerData>> servers_;
};
@@ -186,6 +187,8 @@ TEST_F(RoundRobinEnd2endTest, PickFirst) {
}
}
EXPECT_TRUE(found);
+ // Check LB policy name for the channel.
+ EXPECT_EQ("pick_first", channel_->GetLoadBalancingPolicyName());
}
TEST_F(RoundRobinEnd2endTest, RoundRobin) {
@@ -198,6 +201,8 @@ TEST_F(RoundRobinEnd2endTest, RoundRobin) {
for (size_t i = 0; i < servers_.size(); ++i) {
EXPECT_EQ(1, servers_[i]->service_.request_count());
}
+ // Check LB policy name for the channel.
+ EXPECT_EQ("round_robin", channel_->GetLoadBalancingPolicyName());
}
} // namespace
diff --git a/test/cpp/end2end/test_service_impl.cc b/test/cpp/end2end/test_service_impl.cc
index 2de344efd5..001047778d 100644
--- a/test/cpp/end2end/test_service_impl.cc
+++ b/test/cpp/end2end/test_service_impl.cc
@@ -194,7 +194,7 @@ Status TestServiceImpl::RequestStream(ServerContext* context,
return Status::CANCELLED;
}
- std::thread* server_try_cancel_thd = NULL;
+ std::thread* server_try_cancel_thd = nullptr;
if (server_try_cancel == CANCEL_DURING_PROCESSING) {
server_try_cancel_thd =
new std::thread(&TestServiceImpl::ServerTryCancel, this, context);
@@ -212,7 +212,7 @@ Status TestServiceImpl::RequestStream(ServerContext* context,
}
gpr_log(GPR_INFO, "Read: %d messages", num_msgs_read);
- if (server_try_cancel_thd != NULL) {
+ if (server_try_cancel_thd != nullptr) {
server_try_cancel_thd->join();
delete server_try_cancel_thd;
return Status::CANCELLED;
@@ -248,7 +248,7 @@ Status TestServiceImpl::ResponseStream(ServerContext* context,
}
EchoResponse response;
- std::thread* server_try_cancel_thd = NULL;
+ std::thread* server_try_cancel_thd = nullptr;
if (server_try_cancel == CANCEL_DURING_PROCESSING) {
server_try_cancel_thd =
new std::thread(&TestServiceImpl::ServerTryCancel, this, context);
@@ -259,7 +259,7 @@ Status TestServiceImpl::ResponseStream(ServerContext* context,
writer->Write(response);
}
- if (server_try_cancel_thd != NULL) {
+ if (server_try_cancel_thd != nullptr) {
server_try_cancel_thd->join();
delete server_try_cancel_thd;
return Status::CANCELLED;
@@ -295,7 +295,7 @@ Status TestServiceImpl::BidiStream(
return Status::CANCELLED;
}
- std::thread* server_try_cancel_thd = NULL;
+ std::thread* server_try_cancel_thd = nullptr;
if (server_try_cancel == CANCEL_DURING_PROCESSING) {
server_try_cancel_thd =
new std::thread(&TestServiceImpl::ServerTryCancel, this, context);
@@ -307,7 +307,7 @@ Status TestServiceImpl::BidiStream(
stream->Write(response);
}
- if (server_try_cancel_thd != NULL) {
+ if (server_try_cancel_thd != nullptr) {
server_try_cancel_thd->join();
delete server_try_cancel_thd;
return Status::CANCELLED;
diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc
index fe5a219eed..d353f9894b 100644
--- a/test/cpp/end2end/thread_stress_test.cc
+++ b/test/cpp/end2end/thread_stress_test.cc
@@ -232,19 +232,19 @@ class CommonStressTestSyncServer : public CommonStressTest<TestServiceImpl> {
class CommonStressTestAsyncServer
: public CommonStressTest<grpc::testing::EchoTestService::AsyncService> {
public:
+ CommonStressTestAsyncServer() : contexts_(kNumAsyncServerThreads * 100) {}
void SetUp() override {
shutting_down_ = false;
ServerBuilder builder;
SetUpStart(&builder, &service_);
cq_ = builder.AddCompletionQueue();
SetUpEnd(&builder);
- contexts_ = new Context[kNumAsyncServerThreads * 100];
for (int i = 0; i < kNumAsyncServerThreads * 100; i++) {
RefreshContext(i);
}
for (int i = 0; i < kNumAsyncServerThreads; i++) {
- server_threads_.push_back(
- new std::thread(&CommonStressTestAsyncServer::ProcessRpcs, this));
+ server_threads_.emplace_back(&CommonStressTestAsyncServer::ProcessRpcs,
+ this);
}
}
void TearDown() override {
@@ -256,8 +256,7 @@ class CommonStressTestAsyncServer
}
for (int i = 0; i < kNumAsyncServerThreads; i++) {
- server_threads_[i]->join();
- delete server_threads_[i];
+ server_threads_[i].join();
}
void* ignored_tag;
@@ -265,7 +264,6 @@ class CommonStressTestAsyncServer
while (cq_->Next(&ignored_tag, &ignored_ok))
;
TearDownEnd();
- delete[] contexts_;
}
private:
@@ -311,12 +309,13 @@ class CommonStressTestAsyncServer
response_writer;
EchoRequest recv_request;
enum { READY, DONE } state;
- } * contexts_;
+ };
+ std::vector<Context> contexts_;
::grpc::testing::EchoTestService::AsyncService service_;
std::unique_ptr<ServerCompletionQueue> cq_;
bool shutting_down_;
std::mutex mu_;
- std::vector<std::thread*> server_threads_;
+ std::vector<std::thread> server_threads_;
};
template <class Common>
@@ -353,14 +352,12 @@ typedef ::testing::Types<CommonStressTestSyncServer,
TYPED_TEST_CASE(End2endTest, CommonTypes);
TYPED_TEST(End2endTest, ThreadStress) {
this->common_.ResetStub();
- std::vector<std::thread*> threads;
+ std::vector<std::thread> threads;
for (int i = 0; i < kNumThreads; ++i) {
- threads.push_back(
- new std::thread(SendRpc, this->common_.GetStub(), kNumRpcs));
+ threads.emplace_back(SendRpc, this->common_.GetStub(), kNumRpcs);
}
for (int i = 0; i < kNumThreads; ++i) {
- threads[i]->join();
- delete threads[i];
+ threads[i].join();
}
}
@@ -442,26 +439,24 @@ class AsyncClientEnd2endTest : public ::testing::Test {
TYPED_TEST_CASE(AsyncClientEnd2endTest, CommonTypes);
TYPED_TEST(AsyncClientEnd2endTest, ThreadStress) {
this->common_.ResetStub();
- std::vector<std::thread *> send_threads, completion_threads;
+ std::vector<std::thread> send_threads, completion_threads;
for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
- completion_threads.push_back(new std::thread(
+ completion_threads.emplace_back(
&AsyncClientEnd2endTest_ThreadStress_Test<TypeParam>::AsyncCompleteRpc,
- this));
+ this);
}
for (int i = 0; i < kNumAsyncSendThreads; ++i) {
- send_threads.push_back(new std::thread(
+ send_threads.emplace_back(
&AsyncClientEnd2endTest_ThreadStress_Test<TypeParam>::AsyncSendRpc,
- this, kNumRpcs));
+ this, kNumRpcs);
}
for (int i = 0; i < kNumAsyncSendThreads; ++i) {
- send_threads[i]->join();
- delete send_threads[i];
+ send_threads[i].join();
}
this->Wait();
for (int i = 0; i < kNumAsyncReceiveThreads; ++i) {
- completion_threads[i]->join();
- delete completion_threads[i];
+ completion_threads[i].join();
}
}
diff --git a/test/cpp/grpclb/grpclb_test.cc b/test/cpp/grpclb/grpclb_test.cc
index 175786332b..fcdcaba6a2 100644
--- a/test/cpp/grpclb/grpclb_test.cc
+++ b/test/cpp/grpclb/grpclb_test.cc
@@ -79,6 +79,9 @@ extern "C" {
// - Test against a non-LB server.
// - Random LB server closing the stream unexpectedly.
// - Test using DNS-resolvable names (localhost?)
+// - Test handling of creation of faulty RR instance by having the LB return a
+// serverlist with non-existent backends after having initially returned a
+// valid one.
//
// Findings from end to end testing to be covered here:
// - Handling of LB servers restart, including reconnection after backing-off
@@ -108,6 +111,7 @@ typedef struct server_fixture {
grpc_completion_queue *cq;
char *servers_hostport;
int port;
+ const char *lb_token_prefix;
gpr_thd_id tid;
int num_calls_serviced;
} server_fixture;
@@ -123,7 +127,8 @@ static void *tag(intptr_t t) { return (void *)t; }
static grpc_slice build_response_payload_slice(
const char *host, int *ports, size_t nports,
- int64_t expiration_interval_secs, int32_t expiration_interval_nanos) {
+ int64_t expiration_interval_secs, int32_t expiration_interval_nanos,
+ const char *token_prefix) {
// server_list {
// servers {
// ip_address: <in_addr/6 bytes of an IP>
@@ -150,15 +155,15 @@ static grpc_slice build_response_payload_slice(
struct in_addr ip4;
GPR_ASSERT(inet_pton(AF_INET, host, &ip4) == 1);
server->set_ip_address(
- grpc::string(reinterpret_cast<const char *>(&ip4), sizeof(ip4)));
+ string(reinterpret_cast<const char *>(&ip4), sizeof(ip4)));
server->set_port(ports[i]);
- // The following long long int cast is meant to work around the
- // disfunctional implementation of std::to_string in gcc 4.4, which doesn't
- // have a version for int but does have one for long long int.
- string token_data = "token" + std::to_string((long long int)ports[i]);
- server->set_load_balance_token(token_data);
+ // Missing tokens are acceptable. Test that path.
+ if (strlen(token_prefix) > 0) {
+ string token_data = token_prefix + std::to_string(ports[i]);
+ server->set_load_balance_token(token_data);
+ }
}
- const grpc::string &enc_resp = response.SerializeAsString();
+ const string &enc_resp = response.SerializeAsString();
return grpc_slice_from_copied_buffer(enc_resp.data(), enc_resp.size());
}
@@ -250,14 +255,14 @@ static void start_lb_server(server_fixture *sf, int *ports, size_t nports,
for (int i = 0; i < 2; i++) {
if (i == 0) {
// First half of the ports.
- response_payload_slice =
- build_response_payload_slice("127.0.0.1", ports, nports / 2, -1, -1);
+ response_payload_slice = build_response_payload_slice(
+ "127.0.0.1", ports, nports / 2, -1, -1, sf->lb_token_prefix);
} else {
// Second half of the ports.
sleep_ms(update_delay_ms);
- response_payload_slice =
- build_response_payload_slice("127.0.0.1", ports + (nports / 2),
- (nports + 1) / 2 /* ceil */, -1, -1);
+ response_payload_slice = build_response_payload_slice(
+ "127.0.0.1", ports + (nports / 2), (nports + 1) / 2 /* ceil */, -1,
+ -1, "" /* this half doesn't get to receive an LB token */);
}
response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1);
@@ -339,11 +344,9 @@ static void start_backend_server(server_fixture *sf) {
return;
}
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
-
- // The following long long int cast is meant to work around the
- // disfunctional implementation of std::to_string in gcc 4.4, which doesn't
- // have a version for int but does have one for long long int.
- string expected_token = "token" + std::to_string((long long int)sf->port);
+ const string expected_token =
+ strlen(sf->lb_token_prefix) == 0 ? "" : sf->lb_token_prefix +
+ std::to_string(sf->port);
GPR_ASSERT(contains_metadata(&request_metadata_recv, "lb-token",
expected_token.c_str()));
@@ -520,6 +523,7 @@ static void perform_request(client_fixture *cf) {
CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
cq_verify(cqv);
+ gpr_log(GPR_INFO, "Client after sending msg %d / 4", i + 1);
GPR_ASSERT(byte_buffer_eq_string(response_payload_recv, PAYLOAD));
grpc_byte_buffer_destroy(request_payload);
@@ -541,16 +545,17 @@ static void perform_request(client_fixture *cf) {
cq_verify(cqv);
peer = grpc_call_get_peer(c);
gpr_log(GPR_INFO, "Client DONE WITH SERVER %s ", peer);
- gpr_free(peer);
grpc_call_destroy(c);
- cq_verify_empty_timeout(cqv, 1);
+ cq_verify_empty_timeout(cqv, 1 /* seconds */);
cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
gpr_free(details);
+ gpr_log(GPR_INFO, "Client call (peer %s) DESTROYED.", peer);
+ gpr_free(peer);
}
static void setup_client(const char *server_hostport, client_fixture *cf) {
@@ -626,6 +631,7 @@ static void fork_lb_server(void *arg) {
tf->lb_server_update_delay_ms);
}
+#define LB_TOKEN_PREFIX "token"
static test_fixture setup_test_fixture(int lb_server_update_delay_ms) {
test_fixture tf;
memset(&tf, 0, sizeof(tf));
@@ -635,11 +641,18 @@ static test_fixture setup_test_fixture(int lb_server_update_delay_ms) {
gpr_thd_options_set_joinable(&options);
for (int i = 0; i < NUM_BACKENDS; ++i) {
+ // Only the first half of the servers expect an LB token.
+ if (i < NUM_BACKENDS / 2) {
+ tf.lb_backends[i].lb_token_prefix = LB_TOKEN_PREFIX;
+ } else {
+ tf.lb_backends[i].lb_token_prefix = "";
+ }
setup_server("127.0.0.1", &tf.lb_backends[i]);
gpr_thd_new(&tf.lb_backends[i].tid, fork_backend_server, &tf.lb_backends[i],
&options);
}
+ tf.lb_server.lb_token_prefix = LB_TOKEN_PREFIX;
setup_server("127.0.0.1", &tf.lb_server);
gpr_thd_new(&tf.lb_server.tid, fork_lb_server, &tf.lb_server, &options);
@@ -686,39 +699,42 @@ static test_fixture test_update(int lb_server_update_delay_ms) {
TEST(GrpclbTest, Updates) {
grpc::test_fixture tf_result;
- // Clients take a bit over one second to complete a call (the last part of the
+ // Clients take at least one second to complete a call (the last part of the
// call sleeps for 1 second while verifying the client's completion queue is
- // empty). Therefore:
+ // empty), more if the system is under load. Therefore:
//
// If the LB server waits 800ms before sending an update, it will arrive
- // before the first client request is done, skipping the second server from
- // batch 1 altogether: the 2nd client request will go to the 1st server of
- // batch 2 (ie, the third one out of the four total servers).
+ // before the first client request finishes, skipping the second server from
+ // batch 1. All subsequent picks will come from the second half of the
+ // backends, those coming in the LB update.
tf_result = grpc::test_update(800);
GPR_ASSERT(tf_result.lb_backends[0].num_calls_serviced == 1);
GPR_ASSERT(tf_result.lb_backends[1].num_calls_serviced == 0);
- GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced == 2);
- GPR_ASSERT(tf_result.lb_backends[3].num_calls_serviced == 1);
+ GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced +
+ tf_result.lb_backends[3].num_calls_serviced >
+ 0);
+ int num_serviced_calls = 0;
+ for (int i = 0; i < 4; i++) {
+ num_serviced_calls += tf_result.lb_backends[i].num_calls_serviced;
+ }
+ GPR_ASSERT(num_serviced_calls == 4);
- // If the LB server waits 1500ms, the update arrives after having picked the
- // 2nd server from batch 1 but before the next pick for the first server of
- // batch 2. All server are used.
- tf_result = grpc::test_update(1500);
- GPR_ASSERT(tf_result.lb_backends[0].num_calls_serviced == 1);
- GPR_ASSERT(tf_result.lb_backends[1].num_calls_serviced == 1);
- GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced == 1);
- GPR_ASSERT(tf_result.lb_backends[3].num_calls_serviced == 1);
-
- // If the LB server waits > 2000ms, the update arrives after the first two
- // request are done and the third pick is performed, which returns, in RR
- // fashion, the 1st server of the 1st update. Therefore, the second server of
- // batch 1 is hit at least one, whereas the first server of batch 2 is never
- // hit.
+ // If the LB server waits 2500ms, the update arrives after two calls and three
+ // picks. The third pick will be the 1st server of the 1st update (RR policy
+ // going around). The fourth and final pick will come from the second LB
+ // update. In any case, the total number of serviced calls must again be equal
+ // to four across all the backends.
tf_result = grpc::test_update(2500);
GPR_ASSERT(tf_result.lb_backends[0].num_calls_serviced >= 1);
- GPR_ASSERT(tf_result.lb_backends[1].num_calls_serviced > 0);
- GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced > 0);
- GPR_ASSERT(tf_result.lb_backends[3].num_calls_serviced == 0);
+ GPR_ASSERT(tf_result.lb_backends[1].num_calls_serviced == 1);
+ GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced +
+ tf_result.lb_backends[3].num_calls_serviced >
+ 0);
+ num_serviced_calls = 0;
+ for (int i = 0; i < 4; i++) {
+ num_serviced_calls += tf_result.lb_backends[i].num_calls_serviced;
+ }
+ GPR_ASSERT(num_serviced_calls == 4);
}
TEST(GrpclbTest, InvalidAddressInServerlist) {}
diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc
new file mode 100644
index 0000000000..6cc780d44a
--- /dev/null
+++ b/test/cpp/microbenchmarks/bm_fullstack.cc
@@ -0,0 +1,450 @@
+/*
+ *
+ * Copyright 2016, 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:
+ *
+ * * 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.
+ *
+ */
+
+/* Benchmark gRPC end2end in various configurations */
+
+#include <sstream>
+
+#include <grpc++/channel.h>
+#include <grpc++/create_channel.h>
+#include <grpc++/impl/grpc_library.h>
+#include <grpc++/security/credentials.h>
+#include <grpc++/security/server_credentials.h>
+#include <grpc++/server.h>
+#include <grpc++/server_builder.h>
+#include <grpc/support/log.h>
+
+extern "C" {
+#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/iomgr/endpoint.h"
+#include "src/core/lib/iomgr/endpoint_pair.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/iomgr/tcp_posix.h"
+#include "src/core/lib/surface/channel.h"
+#include "src/core/lib/surface/completion_queue.h"
+#include "src/core/lib/surface/server.h"
+#include "test/core/util/passthru_endpoint.h"
+#include "test/core/util/port.h"
+}
+#include "src/cpp/client/create_channel_internal.h"
+#include "src/proto/grpc/testing/echo.grpc.pb.h"
+#include "third_party/google_benchmark/include/benchmark/benchmark.h"
+
+namespace grpc {
+namespace testing {
+
+static class InitializeStuff {
+ public:
+ InitializeStuff() {
+ init_lib_.init();
+ rq_ = grpc_resource_quota_create("bm");
+ }
+
+ ~InitializeStuff() { init_lib_.shutdown(); }
+
+ grpc_resource_quota* rq() { return rq_; }
+
+ private:
+ internal::GrpcLibrary init_lib_;
+ grpc_resource_quota* rq_;
+} initialize_stuff;
+
+/*******************************************************************************
+ * FIXTURES
+ */
+
+class FullstackFixture {
+ public:
+ FullstackFixture(Service* service, const grpc::string& address) {
+ ServerBuilder b;
+ b.AddListeningPort(address, InsecureServerCredentials());
+ cq_ = b.AddCompletionQueue(true);
+ b.RegisterService(service);
+ server_ = b.BuildAndStart();
+ channel_ = CreateChannel(address, InsecureChannelCredentials());
+ }
+
+ virtual ~FullstackFixture() {
+ server_->Shutdown();
+ cq_->Shutdown();
+ void* tag;
+ bool ok;
+ while (cq_->Next(&tag, &ok)) {
+ }
+ }
+
+ ServerCompletionQueue* cq() { return cq_.get(); }
+ std::shared_ptr<Channel> channel() { return channel_; }
+
+ private:
+ std::unique_ptr<Server> server_;
+ std::unique_ptr<ServerCompletionQueue> cq_;
+ std::shared_ptr<Channel> channel_;
+};
+
+class TCP : public FullstackFixture {
+ public:
+ TCP(Service* service) : FullstackFixture(service, MakeAddress()) {}
+
+ private:
+ static grpc::string MakeAddress() {
+ int port = grpc_pick_unused_port_or_die();
+ std::stringstream addr;
+ addr << "localhost:" << port;
+ return addr.str();
+ }
+};
+
+class UDS : public FullstackFixture {
+ public:
+ UDS(Service* service) : FullstackFixture(service, MakeAddress()) {}
+
+ private:
+ static grpc::string MakeAddress() {
+ int port = grpc_pick_unused_port_or_die(); // just for a unique id - not a
+ // real port
+ std::stringstream addr;
+ addr << "unix:/tmp/bm_fullstack." << port;
+ return addr.str();
+ }
+};
+
+class EndpointPairFixture {
+ public:
+ EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints) {
+ ServerBuilder b;
+ cq_ = b.AddCompletionQueue(true);
+ b.RegisterService(service);
+ server_ = b.BuildAndStart();
+
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+
+ /* add server endpoint to server_ */
+ {
+ const grpc_channel_args* server_args =
+ grpc_server_get_channel_args(server_->c_server());
+ grpc_transport* transport = grpc_create_chttp2_transport(
+ &exec_ctx, server_args, endpoints.server, 0 /* is_client */);
+
+ grpc_pollset** pollsets;
+ size_t num_pollsets = 0;
+ grpc_server_get_pollsets(server_->c_server(), &pollsets, &num_pollsets);
+
+ for (size_t i = 0; i < num_pollsets; i++) {
+ grpc_endpoint_add_to_pollset(&exec_ctx, endpoints.server, pollsets[i]);
+ }
+
+ grpc_server_setup_transport(&exec_ctx, server_->c_server(), transport,
+ NULL, server_args);
+ grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
+ }
+
+ /* create channel */
+ {
+ ChannelArguments args;
+ args.SetString(GRPC_ARG_DEFAULT_AUTHORITY, "test.authority");
+
+ grpc_channel_args c_args = args.c_channel_args();
+ grpc_transport* transport =
+ grpc_create_chttp2_transport(&exec_ctx, &c_args, endpoints.client, 1);
+ GPR_ASSERT(transport);
+ grpc_channel* channel = grpc_channel_create(
+ &exec_ctx, "target", &c_args, GRPC_CLIENT_DIRECT_CHANNEL, transport);
+ grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
+
+ channel_ = CreateChannelInternal("", channel);
+ }
+
+ grpc_exec_ctx_finish(&exec_ctx);
+ }
+
+ virtual ~EndpointPairFixture() {
+ server_->Shutdown();
+ cq_->Shutdown();
+ void* tag;
+ bool ok;
+ while (cq_->Next(&tag, &ok)) {
+ }
+ }
+
+ ServerCompletionQueue* cq() { return cq_.get(); }
+ std::shared_ptr<Channel> channel() { return channel_; }
+
+ private:
+ std::unique_ptr<Server> server_;
+ std::unique_ptr<ServerCompletionQueue> cq_;
+ std::shared_ptr<Channel> channel_;
+};
+
+class SockPair : public EndpointPairFixture {
+ public:
+ SockPair(Service* service)
+ : EndpointPairFixture(service, grpc_iomgr_create_endpoint_pair(
+ "test", initialize_stuff.rq(), 8192)) {
+ }
+};
+
+class InProcessCHTTP2 : public EndpointPairFixture {
+ public:
+ InProcessCHTTP2(Service* service)
+ : EndpointPairFixture(service, MakeEndpoints()) {}
+
+ private:
+ grpc_endpoint_pair MakeEndpoints() {
+ grpc_endpoint_pair p;
+ grpc_passthru_endpoint_create(&p.client, &p.server, initialize_stuff.rq());
+ return p;
+ }
+};
+
+/*******************************************************************************
+ * CONTEXT MUTATORS
+ */
+
+static const int kPregenerateKeyCount = 100000;
+
+template <class F>
+auto MakeVector(size_t length, F f) -> std::vector<decltype(f())> {
+ std::vector<decltype(f())> out;
+ out.reserve(length);
+ for (size_t i = 0; i < length; i++) {
+ out.push_back(f());
+ }
+ return out;
+}
+
+class NoOpMutator {
+ public:
+ template <class ContextType>
+ NoOpMutator(ContextType* context) {}
+};
+
+template <int length>
+class RandomBinaryMetadata {
+ public:
+ static const grpc::string& Key() { return kKey; }
+
+ static const grpc::string& Value() {
+ return kValues[rand() % kValues.size()];
+ }
+
+ private:
+ static const grpc::string kKey;
+ static const std::vector<grpc::string> kValues;
+
+ static grpc::string GenerateOneString() {
+ grpc::string s;
+ s.reserve(length + 1);
+ for (int i = 0; i < length; i++) {
+ s += (char)rand();
+ }
+ return s;
+ }
+};
+
+template <int length>
+const grpc::string RandomBinaryMetadata<length>::kKey = "foo-bin";
+
+template <int length>
+const std::vector<grpc::string> RandomBinaryMetadata<length>::kValues =
+ MakeVector(kPregenerateKeyCount, GenerateOneString);
+
+template <int length>
+class RandomAsciiMetadata {
+ public:
+ static const grpc::string& Key() { return kKey; }
+
+ static const grpc::string& Value() {
+ return kValues[rand() % kValues.size()];
+ }
+
+ private:
+ static const grpc::string kKey;
+ static const std::vector<grpc::string> kValues;
+
+ static grpc::string GenerateOneString() {
+ grpc::string s;
+ s.reserve(length + 1);
+ for (int i = 0; i < length; i++) {
+ s += (char)(rand() % 26 + 'a');
+ }
+ return s;
+ }
+};
+
+template <int length>
+const grpc::string RandomAsciiMetadata<length>::kKey = "foo";
+
+template <int length>
+const std::vector<grpc::string> RandomAsciiMetadata<length>::kValues =
+ MakeVector(kPregenerateKeyCount, GenerateOneString);
+
+template <class Generator, int kNumKeys>
+class Client_AddMetadata : public NoOpMutator {
+ public:
+ Client_AddMetadata(ClientContext* context) : NoOpMutator(context) {
+ for (int i = 0; i < kNumKeys; i++) {
+ context->AddMetadata(Generator::Key(), Generator::Value());
+ }
+ }
+};
+
+template <class Generator, int kNumKeys>
+class Server_AddInitialMetadata : public NoOpMutator {
+ public:
+ Server_AddInitialMetadata(ServerContext* context) : NoOpMutator(context) {
+ for (int i = 0; i < kNumKeys; i++) {
+ context->AddInitialMetadata(Generator::Key(), Generator::Value());
+ }
+ }
+};
+
+/*******************************************************************************
+ * BENCHMARKING KERNELS
+ */
+
+static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); }
+
+template <class Fixture, class ClientContextMutator, class ServerContextMutator>
+static void BM_UnaryPingPong(benchmark::State& state) {
+ EchoTestService::AsyncService service;
+ std::unique_ptr<Fixture> fixture(new Fixture(&service));
+ EchoRequest send_request;
+ EchoResponse send_response;
+ EchoResponse recv_response;
+ Status recv_status;
+ struct ServerEnv {
+ ServerContext ctx;
+ EchoRequest recv_request;
+ grpc::ServerAsyncResponseWriter<EchoResponse> response_writer;
+ ServerEnv() : response_writer(&ctx) {}
+ };
+ uint8_t server_env_buffer[2 * sizeof(ServerEnv)];
+ ServerEnv* server_env[2] = {
+ reinterpret_cast<ServerEnv*>(server_env_buffer),
+ reinterpret_cast<ServerEnv*>(server_env_buffer + sizeof(ServerEnv))};
+ new (server_env[0]) ServerEnv;
+ new (server_env[1]) ServerEnv;
+ service.RequestEcho(&server_env[0]->ctx, &server_env[0]->recv_request,
+ &server_env[0]->response_writer, fixture->cq(),
+ fixture->cq(), tag(0));
+ service.RequestEcho(&server_env[1]->ctx, &server_env[1]->recv_request,
+ &server_env[1]->response_writer, fixture->cq(),
+ fixture->cq(), tag(1));
+ std::unique_ptr<EchoTestService::Stub> stub(
+ EchoTestService::NewStub(fixture->channel()));
+ while (state.KeepRunning()) {
+ ClientContext cli_ctx;
+ ClientContextMutator cli_ctx_mut(&cli_ctx);
+ std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
+ stub->AsyncEcho(&cli_ctx, send_request, fixture->cq()));
+ void* t;
+ bool ok;
+ GPR_ASSERT(fixture->cq()->Next(&t, &ok));
+ GPR_ASSERT(ok);
+ GPR_ASSERT(t == tag(0) || t == tag(1));
+ intptr_t slot = reinterpret_cast<intptr_t>(t);
+ ServerEnv* senv = server_env[slot];
+ ServerContextMutator svr_ctx_mut(&senv->ctx);
+ senv->response_writer.Finish(send_response, Status::OK, tag(3));
+ response_reader->Finish(&recv_response, &recv_status, tag(4));
+ for (int i = (1 << 3) | (1 << 4); i != 0;) {
+ GPR_ASSERT(fixture->cq()->Next(&t, &ok));
+ GPR_ASSERT(ok);
+ int tagnum = (int)reinterpret_cast<intptr_t>(t);
+ GPR_ASSERT(i & (1 << tagnum));
+ i -= 1 << tagnum;
+ }
+ GPR_ASSERT(recv_status.ok());
+
+ senv->~ServerEnv();
+ senv = new (senv) ServerEnv();
+ service.RequestEcho(&senv->ctx, &senv->recv_request, &senv->response_writer,
+ fixture->cq(), fixture->cq(), tag(slot));
+ }
+ fixture.reset();
+ server_env[0]->~ServerEnv();
+ server_env[1]->~ServerEnv();
+}
+
+/*******************************************************************************
+ * CONFIGURATIONS
+ */
+
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, TCP, NoOpMutator, NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, UDS, NoOpMutator, NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, SockPair, NoOpMutator, NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator, NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomBinaryMetadata<10>, 1>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomBinaryMetadata<31>, 1>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomBinaryMetadata<100>, 1>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomBinaryMetadata<10>, 2>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomBinaryMetadata<31>, 2>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomBinaryMetadata<100>, 2>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator,
+ Server_AddInitialMetadata<RandomBinaryMetadata<10>, 1>);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator,
+ Server_AddInitialMetadata<RandomBinaryMetadata<31>, 1>);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator,
+ Server_AddInitialMetadata<RandomBinaryMetadata<100>, 1>);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomAsciiMetadata<10>, 1>, NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomAsciiMetadata<31>, 1>, NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
+ Client_AddMetadata<RandomAsciiMetadata<100>, 1>,
+ NoOpMutator);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator,
+ Server_AddInitialMetadata<RandomAsciiMetadata<10>, 1>);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator,
+ Server_AddInitialMetadata<RandomAsciiMetadata<31>, 1>);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator,
+ Server_AddInitialMetadata<RandomAsciiMetadata<100>, 1>);
+
+} // namespace testing
+} // namespace grpc
+
+BENCHMARK_MAIN();
diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h
index 3a0c359b4c..fdd78ebb89 100644
--- a/test/cpp/qps/client.h
+++ b/test/cpp/qps/client.h
@@ -163,10 +163,9 @@ class Client {
MaybeStartRequests();
- // avoid std::vector for old compilers that expect a copy constructor
if (reset) {
- Histogram* to_merge = new Histogram[threads_.size()];
- StatusHistogram* to_merge_status = new StatusHistogram[threads_.size()];
+ std::vector<Histogram> to_merge(threads_.size());
+ std::vector<StatusHistogram> to_merge_status(threads_.size());
for (size_t i = 0; i < threads_.size(); i++) {
threads_[i]->BeginSwap(&to_merge[i], &to_merge_status[i]);
@@ -177,8 +176,6 @@ class Client {
latencies.Merge(to_merge[i]);
MergeStatusHistogram(to_merge_status[i], &statuses);
}
- delete[] to_merge;
- delete[] to_merge_status;
timer_result = timer->Mark();
} else {
// merge snapshots of each thread histogram
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 2ec6a5a23b..4032039ea1 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -177,7 +177,6 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
shutdown_state_.emplace_back(new PerThreadShutdownState());
}
- using namespace std::placeholders;
int t = 0;
for (int ch = 0; ch < config.client_channels(); ch++) {
for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc
index a88a24d89c..b1e61865e7 100644
--- a/test/cpp/qps/client_sync.cc
+++ b/test/cpp/qps/client_sync.cc
@@ -138,10 +138,9 @@ class SynchronousUnaryClient final : public SynchronousClient {
class SynchronousStreamingClient final : public SynchronousClient {
public:
SynchronousStreamingClient(const ClientConfig& config)
- : SynchronousClient(config) {
- context_ = new grpc::ClientContext[num_threads_];
- stream_ = new std::unique_ptr<
- grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>[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]);
@@ -161,8 +160,6 @@ class SynchronousStreamingClient final : public SynchronousClient {
}
}
}
- delete[] stream_;
- delete[] context_;
}
bool ThreadFunc(HistogramEntry* entry, size_t thread_idx) override {
@@ -182,8 +179,9 @@ class SynchronousStreamingClient final : public SynchronousClient {
private:
// These are both conceptually std::vector but cannot be for old compilers
// that expect contained classes to support copy constructors
- grpc::ClientContext* context_;
- std::unique_ptr<grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>*
+ std::vector<grpc::ClientContext> context_;
+ std::vector<
+ std::unique_ptr<grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>>
stream_;
};
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index a440341ccf..ea0b38e8ad 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -125,6 +125,8 @@ static double UserTime(ClientStats s) { return s.time_user(); }
static double ServerWallTime(ServerStats s) { return s.time_elapsed(); }
static double ServerSystemTime(ServerStats s) { return s.time_system(); }
static double ServerUserTime(ServerStats s) { return s.time_user(); }
+static double ServerTotalCpuTime(ServerStats s) { return s.total_cpu_time(); }
+static double ServerIdleCpuTime(ServerStats s) { return s.idle_cpu_time(); }
static int Cores(int n) { return n; }
// Postprocess ScenarioResult and populate result summary.
@@ -149,6 +151,7 @@ static void postprocess_scenario_result(ScenarioResult* result) {
sum(result->server_stats(), ServerWallTime);
auto server_user_time = 100.0 * sum(result->server_stats(), ServerUserTime) /
sum(result->server_stats(), ServerWallTime);
+
auto client_system_time = 100.0 * sum(result->client_stats(), SystemTime) /
sum(result->client_stats(), WallTime);
auto client_user_time = 100.0 * sum(result->client_stats(), UserTime) /
@@ -159,6 +162,18 @@ static void postprocess_scenario_result(ScenarioResult* result) {
result->mutable_summary()->set_client_system_time(client_system_time);
result->mutable_summary()->set_client_user_time(client_user_time);
+ // For Non-linux platform, get_cpu_usage() is not implemented. Thus,
+ // ServerTotalCpuTime and ServerIdleCpuTime are both 0.
+ if (average(result->server_stats(), ServerTotalCpuTime) == 0) {
+ result->mutable_summary()->set_server_cpu_usage(0);
+ } else {
+ auto server_cpu_usage =
+ 100 -
+ 100 * average(result->server_stats(), ServerIdleCpuTime) /
+ average(result->server_stats(), ServerTotalCpuTime);
+ result->mutable_summary()->set_server_cpu_usage(server_cpu_usage);
+ }
+
if (result->request_results_size() > 0) {
int64_t successes = 0;
int64_t failures = 0;
@@ -177,30 +192,6 @@ static void postprocess_scenario_result(ScenarioResult* result) {
}
}
-// Namespace for classes and functions used only in RunScenario
-// Using this rather than local definitions to workaround gcc-4.4 limitations
-// regarding using templates without linkage
-namespace runsc {
-
-// ClientContext allocator
-static ClientContext* AllocContext(list<ClientContext>* contexts) {
- contexts->emplace_back();
- auto context = &contexts->back();
- context->set_wait_for_ready(true);
- return context;
-}
-
-struct ServerData {
- unique_ptr<WorkerService::Stub> stub;
- unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
-};
-
-struct ClientData {
- unique_ptr<WorkerService::Stub> stub;
- unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
-};
-} // namespace runsc
-
std::unique_ptr<ScenarioResult> RunScenario(
const ClientConfig& initial_client_config, size_t num_clients,
const ServerConfig& initial_server_config, size_t num_servers,
@@ -210,6 +201,12 @@ std::unique_ptr<ScenarioResult> RunScenario(
// ClientContext allocations (all are destroyed at scope exit)
list<ClientContext> contexts;
+ auto alloc_context = [](list<ClientContext>* contexts) {
+ contexts->emplace_back();
+ auto context = &contexts->back();
+ context->set_wait_for_ready(true);
+ return context;
+ };
// To be added to the result, containing the final configuration used for
// client and config (including host, etc.)
@@ -262,10 +259,11 @@ std::unique_ptr<ScenarioResult> RunScenario(
workers.resize(num_clients + num_servers);
// Start servers
- using runsc::ServerData;
- // servers is array rather than std::vector to avoid gcc-4.4 issues
- // where class contained in std::vector must have a copy constructor
- auto* servers = new ServerData[num_servers];
+ struct ServerData {
+ unique_ptr<WorkerService::Stub> stub;
+ unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
+ };
+ std::vector<ServerData> servers(num_servers);
for (size_t i = 0; i < num_servers; i++) {
gpr_log(GPR_INFO, "Starting server on %s (worker #%" PRIuPTR ")",
workers[i].c_str(), i);
@@ -309,8 +307,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
ServerArgs args;
*args.mutable_setup() = server_config;
- servers[i].stream =
- servers[i].stub->RunServer(runsc::AllocContext(&contexts));
+ servers[i].stream = servers[i].stub->RunServer(alloc_context(&contexts));
if (!servers[i].stream->Write(args)) {
gpr_log(GPR_ERROR, "Could not write args to server %zu", i);
}
@@ -328,10 +325,11 @@ std::unique_ptr<ScenarioResult> RunScenario(
// Targets are all set by now
result_client_config = client_config;
// Start clients
- using runsc::ClientData;
- // clients is array rather than std::vector to avoid gcc-4.4 issues
- // where class contained in std::vector must have a copy constructor
- auto* clients = new ClientData[num_clients];
+ struct ClientData {
+ unique_ptr<WorkerService::Stub> stub;
+ unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
+ };
+ std::vector<ClientData> clients(num_clients);
size_t channels_allocated = 0;
for (size_t i = 0; i < num_clients; i++) {
const auto& worker = workers[i + num_servers];
@@ -380,8 +378,7 @@ std::unique_ptr<ScenarioResult> RunScenario(
ClientArgs args;
*args.mutable_setup() = per_client_config;
- clients[i].stream =
- clients[i].stub->RunClient(runsc::AllocContext(&contexts));
+ clients[i].stream = clients[i].stub->RunClient(alloc_context(&contexts));
if (!clients[i].stream->Write(args)) {
gpr_log(GPR_ERROR, "Could not write args to client %zu", i);
}
@@ -501,7 +498,6 @@ std::unique_ptr<ScenarioResult> RunScenario(
s.error_message().c_str());
}
}
- delete[] clients;
merged_latencies.FillProto(result->mutable_latencies());
for (std::unordered_map<int, int64_t>::iterator it = merged_statuses.begin();
@@ -544,8 +540,6 @@ std::unique_ptr<ScenarioResult> RunScenario(
}
}
- delete[] servers;
-
postprocess_scenario_result(result.get());
return result;
}
diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py
index e4d9e7ac58..4aa58d2737 100755
--- a/test/cpp/qps/gen_build_yaml.py
+++ b/test/cpp/qps/gen_build_yaml.py
@@ -43,28 +43,38 @@ sys.path.append(run_tests_root)
import performance.scenario_config as scenario_config
-def _scenario_json_string(scenario_json):
+configs_from_yaml = yaml.load(open(os.path.join(os.path.dirname(sys.argv[0]), '../../../build.yaml')))['configs'].keys()
+
+def mutate_scenario(scenario_json, is_tsan):
# tweak parameters to get fast test times
+ scenario_json = dict(scenario_json)
scenario_json['warmup_seconds'] = 0
scenario_json['benchmark_seconds'] = 1
- scenarios_json = {'scenarios': [scenario_config.remove_nonproto_fields(scenario_json)]}
+ outstanding_rpcs_divisor = 1
+ if is_tsan and (
+ scenario_json['client_config']['client_type'] == 'SYNC_CLIENT' or
+ scenario_json['server_config']['server_type'] == 'SYNC_SERVER'):
+ outstanding_rpcs_divisor = 10
+ scenario_json['client_config']['outstanding_rpcs_per_channel'] = max(1,
+ int(scenario_json['client_config']['outstanding_rpcs_per_channel'] / outstanding_rpcs_divisor))
+ return scenario_json
+
+def _scenario_json_string(scenario_json, is_tsan):
+ scenarios_json = {'scenarios': [scenario_config.remove_nonproto_fields(mutate_scenario(scenario_json, is_tsan))]}
return json.dumps(scenarios_json)
-def threads_of_type(scenario_json, path):
- d = scenario_json
- for el in path.split('/'):
- if el not in d:
- return 0
- d = d[el]
- return d
+def threads_required(scenario_json, where, is_tsan):
+ scenario_json = mutate_scenario(scenario_json, is_tsan)
+ if scenario_json['%s_config' % where]['%s_type' % where] == 'ASYNC_%s' % where.upper():
+ return scenario_json['%s_config' % where].get('async_%s_threads' % where, 0)
+ return scenario_json['client_config']['outstanding_rpcs_per_channel'] * scenario_json['client_config']['client_channels']
-def guess_cpu(scenario_json):
- client = threads_of_type(scenario_json, 'client_config/async_client_threads')
- server = threads_of_type(scenario_json, 'server_config/async_server_threads')
+def guess_cpu(scenario_json, is_tsan):
+ client = threads_required(scenario_json, 'client', is_tsan)
+ server = threads_required(scenario_json, 'server', is_tsan)
# make an arbitrary guess if set to auto-detect
# about the size of the jenkins instances we have for unit tests
- if client == 0: client = 8
- if server == 0: server = 8
+ if client == 0 or server == 0: return 'capacity'
return (scenario_json['num_clients'] * client +
scenario_json['num_servers'] * server)
@@ -73,15 +83,32 @@ print yaml.dump({
{
'name': 'json_run_localhost',
'shortname': 'json_run_localhost:%s' % scenario_json['name'],
- 'args': ['--scenarios_json', _scenario_json_string(scenario_json)],
+ 'args': ['--scenarios_json', _scenario_json_string(scenario_json, False)],
+ 'ci_platforms': ['linux'],
+ 'platforms': ['linux'],
+ 'flaky': False,
+ 'language': 'c++',
+ 'boringssl': True,
+ 'defaults': 'boringssl',
+ 'cpu_cost': guess_cpu(scenario_json, False),
+ 'exclude_configs': ['tsan'],
+ 'timeout_seconds': 6*60
+ }
+ for scenario_json in scenario_config.CXXLanguage().scenarios()
+ if 'scalable' in scenario_json.get('CATEGORIES', [])
+ ] + [
+ {
+ 'name': 'json_run_localhost',
+ 'shortname': 'json_run_localhost:%s' % scenario_json['name'],
+ 'args': ['--scenarios_json', _scenario_json_string(scenario_json, True)],
'ci_platforms': ['linux'],
'platforms': ['linux'],
'flaky': False,
'language': 'c++',
'boringssl': True,
'defaults': 'boringssl',
- 'cpu_cost': guess_cpu(scenario_json),
- 'exclude_configs': [],
+ 'cpu_cost': guess_cpu(scenario_json, True),
+ 'exclude_configs': sorted(c for c in configs_from_yaml if c != 'tsan'),
'timeout_seconds': 6*60
}
for scenario_json in scenario_config.CXXLanguage().scenarios()
diff --git a/test/cpp/qps/json_run_localhost.cc b/test/cpp/qps/json_run_localhost.cc
index 74e40fbf1a..b7b2553f12 100644
--- a/test/cpp/qps/json_run_localhost.cc
+++ b/test/cpp/qps/json_run_localhost.cc
@@ -31,7 +31,11 @@
*
*/
+#include <signal.h>
+#include <string.h>
+
#include <memory>
+#include <mutex>
#include <sstream>
#include <string>
@@ -43,6 +47,11 @@
using grpc::SubProcess;
+constexpr auto kNumWorkers = 2;
+
+static SubProcess* g_driver;
+static SubProcess* g_workers[kNumWorkers];
+
template <class T>
std::string as_string(const T& val) {
std::ostringstream out;
@@ -50,9 +59,38 @@ std::string as_string(const T& val) {
return out.str();
}
+static void sighandler(int sig) {
+ const int errno_saved = errno;
+ if (g_driver != NULL) g_driver->Interrupt();
+ for (int i = 0; i < kNumWorkers; ++i) {
+ if (g_workers[i]) g_workers[i]->Interrupt();
+ }
+ errno = errno_saved;
+}
+
+static void register_sighandler() {
+ struct sigaction act;
+ memset(&act, 0, sizeof(act));
+ act.sa_handler = sighandler;
+
+ sigaction(SIGINT, &act, NULL);
+ sigaction(SIGTERM, &act, NULL);
+}
+
+static void LogStatus(int status, const char* label) {
+ if (WIFEXITED(status)) {
+ gpr_log(GPR_INFO, "%s: subprocess exited with status %d", label,
+ WEXITSTATUS(status));
+ } else if (WIFSIGNALED(status)) {
+ gpr_log(GPR_INFO, "%s: subprocess terminated with signal %d", label,
+ WTERMSIG(status));
+ } else {
+ gpr_log(GPR_INFO, "%s: unknown subprocess status: %d", label, status);
+ }
+}
+
int main(int argc, char** argv) {
- typedef std::unique_ptr<SubProcess> SubProcessPtr;
- std::vector<SubProcessPtr> jobs;
+ register_sighandler();
std::string my_bin = argv[0];
std::string bin_dir = my_bin.substr(0, my_bin.rfind('/'));
@@ -60,11 +98,11 @@ int main(int argc, char** argv) {
std::ostringstream env;
bool first = true;
- for (int i = 0; i < 2; i++) {
- auto port = grpc_pick_unused_port_or_die();
+ for (int i = 0; i < kNumWorkers; i++) {
+ const auto port = grpc_pick_unused_port_or_die();
std::vector<std::string> args = {bin_dir + "/qps_worker", "-driver_port",
as_string(port)};
- jobs.emplace_back(new SubProcess(args));
+ g_workers[i] = new SubProcess(args);
if (!first) env << ",";
env << "localhost:" << port;
first = false;
@@ -75,12 +113,27 @@ int main(int argc, char** argv) {
for (int i = 1; i < argc; i++) {
args.push_back(argv[i]);
}
- GPR_ASSERT(SubProcess(args).Join() == 0);
- for (auto it = jobs.begin(); it != jobs.end(); ++it) {
- (*it)->Interrupt();
+ g_driver = new SubProcess(args);
+ const int driver_join_status = g_driver->Join();
+ if (driver_join_status != 0) {
+ LogStatus(driver_join_status, "driver");
}
- for (auto it = jobs.begin(); it != jobs.end(); ++it) {
- (*it)->Join();
+ for (int i = 0; i < kNumWorkers; ++i) {
+ if (g_workers[i]) g_workers[i]->Interrupt();
}
+
+ for (int i = 0; i < kNumWorkers; ++i) {
+ if (g_workers[i]) {
+ const int worker_status = g_workers[i]->Join();
+ if (worker_status != 0) {
+ LogStatus(worker_status, "worker");
+ }
+ }
+ }
+
+ delete g_driver;
+ g_driver = NULL;
+ for (int i = 0; i < kNumWorkers; ++i) delete g_workers[i];
+ GPR_ASSERT(driver_join_status == 0);
}
diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc
index 1524ebbc38..31b5917fb7 100644
--- a/test/cpp/qps/qps_json_driver.cc
+++ b/test/cpp/qps/qps_json_driver.cc
@@ -49,10 +49,111 @@ DEFINE_string(scenarios_file, "",
DEFINE_string(scenarios_json, "",
"JSON string containing an array of Scenario objects");
DEFINE_bool(quit, false, "Quit the workers");
+DEFINE_string(search_param, "",
+ "The parameter, whose value is to be searched for to achieve "
+ "targeted cpu load. For now, we have 'offered_load'. Later, "
+ "'num_channels', 'num_outstanding_requests', etc. shall be "
+ "added.");
+DEFINE_double(
+ initial_search_value, 0.0,
+ "initial parameter value to start the search with (i.e. lower bound)");
+DEFINE_double(targeted_cpu_load, 70.0,
+ "Targeted cpu load (unit: %, range [0,100])");
+DEFINE_double(stride, 1,
+ "Defines each stride of the search. The larger the stride is, "
+ "the coarser the result will be, but will also be faster.");
+DEFINE_double(error_tolerance, 0.01,
+ "Defines threshold for stopping the search. When current search "
+ "range is narrower than the error_tolerance computed range, we "
+ "stop the search.");
namespace grpc {
namespace testing {
+static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario,
+ bool* success) {
+ std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
+ auto result =
+ RunScenario(scenario.client_config(), scenario.num_clients(),
+ scenario.server_config(), scenario.num_servers(),
+ scenario.warmup_seconds(), scenario.benchmark_seconds(),
+ scenario.spawn_local_worker_count());
+
+ // Amend the result with scenario config. Eventually we should adjust
+ // RunScenario contract so we don't need to touch the result here.
+ result->mutable_scenario()->CopyFrom(scenario);
+
+ GetReporter()->ReportQPS(*result);
+ GetReporter()->ReportQPSPerCore(*result);
+ GetReporter()->ReportLatency(*result);
+ GetReporter()->ReportTimes(*result);
+ GetReporter()->ReportCpuUsage(*result);
+
+ for (int i = 0; *success && i < result->client_success_size(); i++) {
+ *success = result->client_success(i);
+ }
+ for (int i = 0; *success && i < result->server_success_size(); i++) {
+ *success = result->server_success(i);
+ }
+
+ return result;
+}
+
+static double GetCpuLoad(Scenario* scenario, double offered_load,
+ bool* success) {
+ scenario->mutable_client_config()
+ ->mutable_load_params()
+ ->mutable_poisson()
+ ->set_offered_load(offered_load);
+ auto result = RunAndReport(*scenario, success);
+ return result->summary().server_cpu_usage();
+}
+
+static double BinarySearch(Scenario* scenario, double targeted_cpu_load,
+ double low, double high, bool* success) {
+ while (low <= high * (1 - FLAGS_error_tolerance)) {
+ double mid = low + (high - low) / 2;
+ double current_cpu_load = GetCpuLoad(scenario, mid, success);
+ gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f", mid);
+ if (!*success) {
+ gpr_log(GPR_ERROR, "Client/Server Failure");
+ break;
+ }
+ if (targeted_cpu_load <= current_cpu_load) {
+ high = mid - FLAGS_stride;
+ } else {
+ low = mid + FLAGS_stride;
+ }
+ }
+
+ return low;
+}
+
+static double SearchOfferedLoad(double initial_offered_load,
+ double targeted_cpu_load, Scenario* scenario,
+ bool* success) {
+ std::cerr << "RUNNING SCENARIO: " << scenario->name() << "\n";
+ double current_offered_load = initial_offered_load;
+ double current_cpu_load = GetCpuLoad(scenario, current_offered_load, success);
+ if (current_cpu_load > targeted_cpu_load) {
+ gpr_log(GPR_ERROR, "Initial offered load too high");
+ return -1;
+ }
+
+ while (*success && (current_cpu_load < targeted_cpu_load)) {
+ current_offered_load *= 2;
+ current_cpu_load = GetCpuLoad(scenario, current_offered_load, success);
+ gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f",
+ current_offered_load);
+ }
+
+ double targeted_offered_load =
+ BinarySearch(scenario, targeted_cpu_load, current_offered_load / 2,
+ current_offered_load, success);
+
+ return targeted_offered_load;
+}
+
static bool QpsDriver() {
grpc::string json;
@@ -68,11 +169,11 @@ static bool QpsDriver() {
if (scfile) {
// Read the json data from disk
- FILE *json_file = fopen(FLAGS_scenarios_file.c_str(), "r");
+ FILE* json_file = fopen(FLAGS_scenarios_file.c_str(), "r");
GPR_ASSERT(json_file != NULL);
fseek(json_file, 0, SEEK_END);
long len = ftell(json_file);
- char *data = new char[len];
+ char* data = new char[len];
fseek(json_file, 0, SEEK_SET);
GPR_ASSERT(len == (long)fread(data, 1, len, json_file));
fclose(json_file);
@@ -93,28 +194,19 @@ static bool QpsDriver() {
GPR_ASSERT(scenarios.scenarios_size() > 0);
for (int i = 0; i < scenarios.scenarios_size(); i++) {
- const Scenario &scenario = scenarios.scenarios(i);
- std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
- auto result =
- RunScenario(scenario.client_config(), scenario.num_clients(),
- scenario.server_config(), scenario.num_servers(),
- scenario.warmup_seconds(), scenario.benchmark_seconds(),
- scenario.spawn_local_worker_count());
-
- // Amend the result with scenario config. Eventually we should adjust
- // RunScenario contract so we don't need to touch the result here.
- result->mutable_scenario()->CopyFrom(scenario);
-
- GetReporter()->ReportQPS(*result);
- GetReporter()->ReportQPSPerCore(*result);
- GetReporter()->ReportLatency(*result);
- GetReporter()->ReportTimes(*result);
-
- for (int i = 0; success && i < result->client_success_size(); i++) {
- success = result->client_success(i);
- }
- for (int i = 0; success && i < result->server_success_size(); i++) {
- success = result->server_success(i);
+ if (FLAGS_search_param == "") {
+ const Scenario& scenario = scenarios.scenarios(i);
+ RunAndReport(scenario, &success);
+ } else {
+ if (FLAGS_search_param == "offered_load") {
+ Scenario* scenario = scenarios.mutable_scenarios(i);
+ double targeted_offered_load =
+ SearchOfferedLoad(FLAGS_initial_search_value,
+ FLAGS_targeted_cpu_load, scenario, &success);
+ gpr_log(GPR_INFO, "targeted_offered_load %f", targeted_offered_load);
+ } else {
+ gpr_log(GPR_ERROR, "Unimplemented search param");
+ }
}
}
return success;
@@ -123,7 +215,7 @@ static bool QpsDriver() {
} // namespace testing
} // namespace grpc
-int main(int argc, char **argv) {
+int main(int argc, char** argv) {
grpc::testing::InitBenchmark(&argc, &argv, true);
bool ok = grpc::testing::QpsDriver();
diff --git a/test/cpp/qps/report.cc b/test/cpp/qps/report.cc
index 41617e968a..7f84816421 100644
--- a/test/cpp/qps/report.cc
+++ b/test/cpp/qps/report.cc
@@ -71,6 +71,12 @@ void CompositeReporter::ReportTimes(const ScenarioResult& result) {
}
}
+void CompositeReporter::ReportCpuUsage(const ScenarioResult& result) {
+ for (size_t i = 0; i < reporters_.size(); ++i) {
+ reporters_[i]->ReportCpuUsage(result);
+ }
+}
+
void GprLogReporter::ReportQPS(const ScenarioResult& result) {
gpr_log(GPR_INFO, "QPS: %.1f", result.summary().qps());
if (result.summary().failed_requests_per_second() > 0) {
@@ -107,6 +113,11 @@ void GprLogReporter::ReportTimes(const ScenarioResult& result) {
result.summary().client_user_time());
}
+void GprLogReporter::ReportCpuUsage(const ScenarioResult& result) {
+ gpr_log(GPR_INFO, "Server CPU usage: %.2f%%",
+ result.summary().server_cpu_usage());
+}
+
void JsonReporter::ReportQPS(const ScenarioResult& result) {
grpc::string json_string =
SerializeJson(result, "type.googleapis.com/grpc.testing.ScenarioResult");
@@ -127,5 +138,9 @@ void JsonReporter::ReportTimes(const ScenarioResult& result) {
// NOP - all reporting is handled by ReportQPS.
}
+void JsonReporter::ReportCpuUsage(const ScenarioResult& result) {
+ // NOP - all reporting is handled by ReportQPS.
+}
+
} // namespace testing
} // namespace grpc
diff --git a/test/cpp/qps/report.h b/test/cpp/qps/report.h
index 9dc259e95a..faf87ff060 100644
--- a/test/cpp/qps/report.h
+++ b/test/cpp/qps/report.h
@@ -70,6 +70,9 @@ class Reporter {
/** Reports system and user time for client and server systems. */
virtual void ReportTimes(const ScenarioResult& result) = 0;
+ /** Reports server cpu usage. */
+ virtual void ReportCpuUsage(const ScenarioResult& result) = 0;
+
private:
const string name_;
};
@@ -86,6 +89,7 @@ class CompositeReporter : public Reporter {
void ReportQPSPerCore(const ScenarioResult& result) override;
void ReportLatency(const ScenarioResult& result) override;
void ReportTimes(const ScenarioResult& result) override;
+ void ReportCpuUsage(const ScenarioResult& result) override;
private:
std::vector<std::unique_ptr<Reporter> > reporters_;
@@ -101,6 +105,7 @@ class GprLogReporter : public Reporter {
void ReportQPSPerCore(const ScenarioResult& result) override;
void ReportLatency(const ScenarioResult& result) override;
void ReportTimes(const ScenarioResult& result) override;
+ void ReportCpuUsage(const ScenarioResult& result) override;
};
/** Dumps the report to a JSON file. */
@@ -114,6 +119,7 @@ class JsonReporter : public Reporter {
void ReportQPSPerCore(const ScenarioResult& result) override;
void ReportLatency(const ScenarioResult& result) override;
void ReportTimes(const ScenarioResult& result) override;
+ void ReportCpuUsage(const ScenarioResult& result) override;
const string report_file_;
};
diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h
index e8bc396696..c3d18e5789 100644
--- a/test/cpp/qps/server.h
+++ b/test/cpp/qps/server.h
@@ -75,6 +75,8 @@ class Server {
stats.set_time_elapsed(timer_result.wall);
stats.set_time_system(timer_result.system);
stats.set_time_user(timer_result.user);
+ stats.set_total_cpu_time(timer_result.total_cpu_time);
+ stats.set_idle_cpu_time(timer_result.idle_cpu_time);
return stats;
}
diff --git a/test/cpp/qps/usage_timer.cc b/test/cpp/qps/usage_timer.cc
index ff595b2ba0..c6697fbdfd 100644
--- a/test/cpp/qps/usage_timer.cc
+++ b/test/cpp/qps/usage_timer.cc
@@ -33,10 +33,14 @@
#include "test/cpp/qps/usage_timer.h"
+#include <fstream>
+#include <sstream>
+#include <string>
+
+#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <sys/resource.h>
#include <sys/time.h>
-
UsageTimer::UsageTimer() : start_(Sample()) {}
double UsageTimer::Now() {
@@ -48,6 +52,27 @@ static double time_double(struct timeval* tv) {
return tv->tv_sec + 1e-6 * tv->tv_usec;
}
+static void get_cpu_usage(unsigned long long* total_cpu_time,
+ unsigned long long* idle_cpu_time) {
+#ifdef __linux__
+ std::ifstream proc_stat("/proc/stat");
+ proc_stat.ignore(5);
+ std::string cpu_time_str;
+ std::string first_line;
+ std::getline(proc_stat, first_line);
+ std::stringstream first_line_s(first_line);
+ for (int i = 0; i < 10; ++i) {
+ std::getline(first_line_s, cpu_time_str, ' ');
+ *total_cpu_time += std::stol(cpu_time_str);
+ if (i == 3) {
+ *idle_cpu_time = std::stol(cpu_time_str);
+ }
+ }
+#else
+ gpr_log(GPR_INFO, "get_cpu_usage(): Non-linux platform is not supported.");
+#endif
+}
+
UsageTimer::Result UsageTimer::Sample() {
struct rusage usage;
struct timeval tv;
@@ -58,6 +83,9 @@ UsageTimer::Result UsageTimer::Sample() {
r.wall = time_double(&tv);
r.user = time_double(&usage.ru_utime);
r.system = time_double(&usage.ru_stime);
+ r.total_cpu_time = 0;
+ r.idle_cpu_time = 0;
+ get_cpu_usage(&r.total_cpu_time, &r.idle_cpu_time);
return r;
}
@@ -67,5 +95,8 @@ UsageTimer::Result UsageTimer::Mark() const {
r.wall = s.wall - start_.wall;
r.user = s.user - start_.user;
r.system = s.system - start_.system;
+ r.total_cpu_time = s.total_cpu_time - start_.total_cpu_time;
+ r.idle_cpu_time = s.idle_cpu_time - start_.idle_cpu_time;
+
return r;
}
diff --git a/test/cpp/qps/usage_timer.h b/test/cpp/qps/usage_timer.h
index 8343cd6653..0fc1b47996 100644
--- a/test/cpp/qps/usage_timer.h
+++ b/test/cpp/qps/usage_timer.h
@@ -42,6 +42,8 @@ class UsageTimer {
double wall;
double user;
double system;
+ unsigned long long total_cpu_time;
+ unsigned long long idle_cpu_time;
};
Result Mark() const;
diff --git a/test/cpp/util/config_grpc_cli.h b/test/cpp/util/config_grpc_cli.h
index ea8231aa26..ac1e3044b7 100644
--- a/test/cpp/util/config_grpc_cli.h
+++ b/test/cpp/util/config_grpc_cli.h
@@ -77,7 +77,7 @@ namespace compiler {
typedef GRPC_CUSTOM_DISKSOURCETREE DiskSourceTree;
typedef GRPC_CUSTOM_IMPORTER Importer;
typedef GRPC_CUSTOM_MULTIFILEERRORCOLLECTOR MultiFileErrorCollector;
-} // namespace importer
+} // namespace compiler
} // namespace protobuf
} // namespace grpc
diff --git a/test/cpp/util/grpc_tool_test.cc b/test/cpp/util/grpc_tool_test.cc
index 5ab054d04a..1ff8172306 100644
--- a/test/cpp/util/grpc_tool_test.cc
+++ b/test/cpp/util/grpc_tool_test.cc
@@ -112,8 +112,6 @@ size_t ArraySize(T& a) {
static_cast<size_t>(!(sizeof(a) % sizeof(*(a)))));
}
-} // namespame
-
class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
public:
Status Echo(ServerContext* context, const EchoRequest* request,
@@ -132,6 +130,8 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
}
};
+} // namespace
+
class GrpcToolTest : public ::testing::Test {
protected:
GrpcToolTest() {}
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-05 23:16:13 +0000