diff options
Diffstat (limited to 'test/cpp')
53 files changed, 1004 insertions, 323 deletions
diff --git a/test/cpp/codegen/BUILD b/test/cpp/codegen/BUILD index 984440e455..4780800645 100644 --- a/test/cpp/codegen/BUILD +++ b/test/cpp/codegen/BUILD @@ -16,6 +16,13 @@ licenses(["notice"]) # Apache v2 load("//bazel:grpc_build_system.bzl", "grpc_cc_test") +package( + features = [ + "-layering_check", + "-parse_headers", + ], +) + grpc_cc_test( name = "codegen_test_full", srcs = ["codegen_test_full.cc"], diff --git a/test/cpp/codegen/codegen_test_full.cc b/test/cpp/codegen/codegen_test_full.cc index 2eacc99d82..98792bde04 100644 --- a/test/cpp/codegen/codegen_test_full.cc +++ b/test/cpp/codegen/codegen_test_full.cc @@ -27,8 +27,8 @@ class CodegenTestFull : public ::testing::Test {}; TEST_F(CodegenTestFull, Init) { grpc::CompletionQueue cq; - void* tag; - bool ok; + void* tag = nullptr; + bool ok = false; cq.AsyncNext(&tag, &ok, gpr_time_0(GPR_CLOCK_REALTIME)); ASSERT_FALSE(ok); } diff --git a/test/cpp/common/BUILD b/test/cpp/common/BUILD index ae287a4c93..bd1173322a 100644 --- a/test/cpp/common/BUILD +++ b/test/cpp/common/BUILD @@ -16,6 +16,13 @@ licenses(["notice"]) # Apache v2 load("//bazel:grpc_build_system.bzl", "grpc_cc_test") +package( + features = [ + "-layering_check", + "-parse_headers", + ], +) + grpc_cc_test( name = "alarm_cpp_test", srcs = ["alarm_cpp_test.cc"], diff --git a/test/cpp/end2end/BUILD b/test/cpp/end2end/BUILD index ac742c2921..27c5492c17 100644 --- a/test/cpp/end2end/BUILD +++ b/test/cpp/end2end/BUILD @@ -16,10 +16,17 @@ licenses(["notice"]) # Apache v2 load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_test") -package(default_visibility=["//visibility:public"]) # Allows external users to implement end2end tests. +package( + default_visibility=["//visibility:public"], # Allows external users to implement end2end tests. + features = [ + "-layering_check", + "-parse_headers", + ], +) grpc_cc_library( name = "test_service_impl", + testonly = True, srcs = ["test_service_impl.cc"], hdrs = ["test_service_impl.h"], deps = [ diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index a2a6e36709..7cb7b262de 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -212,14 +212,16 @@ class ServerBuilderSyncPluginDisabler : public ::grpc::ServerBuilderOption { class TestScenario { public: - TestScenario(bool non_block, const grpc::string& creds_type, bool hcs, - const grpc::string& content) + TestScenario(bool non_block, bool inproc_stub, const grpc::string& creds_type, + bool hcs, const grpc::string& content) : disable_blocking(non_block), + inproc(inproc_stub), health_check_service(hcs), credentials_type(creds_type), message_content(content) {} void Log() const; bool disable_blocking; + bool inproc; bool health_check_service; // Although the below grpc::string's are logically const, we can't declare // them const because of a limitation in the way old compilers (e.g., gcc-4.4) @@ -232,6 +234,7 @@ static std::ostream& operator<<(std::ostream& out, const TestScenario& scenario) { return out << "TestScenario{disable_blocking=" << (scenario.disable_blocking ? "true" : "false") + << ", inproc=" << (scenario.inproc ? "true" : "false") << ", credentials='" << scenario.credentials_type << ", health_check_service=" << (scenario.health_check_service ? "true" : "false") @@ -294,7 +297,9 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> { auto channel_creds = GetCredentialsProvider()->GetChannelCredentials( GetParam().credentials_type, &args); std::shared_ptr<Channel> channel = - CreateCustomChannel(server_address_.str(), channel_creds, args); + !(GetParam().inproc) + ? CreateCustomChannel(server_address_.str(), channel_creds, args) + : server_->InProcessChannel(args); stub_ = grpc::testing::EchoTestService::NewStub(channel); } @@ -512,7 +517,7 @@ TEST_P(AsyncEnd2endTest, SimpleClientStreamingWithCoalescingApi) { // up until server read is initiated. For write of send_request smaller than // the flow control window size, the request can take the free ride with // initial metadata due to coalescing, thus write tag:3 will come up here. - if (GetParam().message_content.length() < 65536) { + if (GetParam().message_content.length() < 65536 || GetParam().inproc) { Verifier(GetParam().disable_blocking) .Expect(2, true) .Expect(3, true) @@ -523,7 +528,7 @@ TEST_P(AsyncEnd2endTest, SimpleClientStreamingWithCoalescingApi) { srv_stream.Read(&recv_request, tag(4)); - if (GetParam().message_content.length() < 65536) { + if (GetParam().message_content.length() < 65536 || GetParam().inproc) { Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get()); } else { Verifier(GetParam().disable_blocking) @@ -807,7 +812,7 @@ TEST_P(AsyncEnd2endTest, SimpleBidiStreamingWithCoalescingApiWAF) { // up until server read is initiated. For write of send_request smaller than // the flow control window size, the request can take the free ride with // initial metadata due to coalescing, thus write tag:3 will come up here. - if (GetParam().message_content.length() < 65536) { + if (GetParam().message_content.length() < 65536 || GetParam().inproc) { Verifier(GetParam().disable_blocking) .Expect(2, true) .Expect(3, true) @@ -818,7 +823,7 @@ TEST_P(AsyncEnd2endTest, SimpleBidiStreamingWithCoalescingApiWAF) { srv_stream.Read(&recv_request, tag(4)); - if (GetParam().message_content.length() < 65536) { + if (GetParam().message_content.length() < 65536 || GetParam().inproc) { Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get()); } else { Verifier(GetParam().disable_blocking) @@ -875,7 +880,7 @@ TEST_P(AsyncEnd2endTest, SimpleBidiStreamingWithCoalescingApiWL) { // up until server read is initiated. For write of send_request smaller than // the flow control window size, the request can take the free ride with // initial metadata due to coalescing, thus write tag:3 will come up here. - if (GetParam().message_content.length() < 65536) { + if (GetParam().message_content.length() < 65536 || GetParam().inproc) { Verifier(GetParam().disable_blocking) .Expect(2, true) .Expect(3, true) @@ -886,7 +891,7 @@ TEST_P(AsyncEnd2endTest, SimpleBidiStreamingWithCoalescingApiWL) { srv_stream.Read(&recv_request, tag(4)); - if (GetParam().message_content.length() < 65536) { + if (GetParam().message_content.length() < 65536 || GetParam().inproc) { Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get()); } else { Verifier(GetParam().disable_blocking) @@ -1223,7 +1228,9 @@ TEST_P(AsyncEnd2endTest, UnimplementedRpc) { auto channel_creds = GetCredentialsProvider()->GetChannelCredentials( GetParam().credentials_type, &args); std::shared_ptr<Channel> channel = - CreateCustomChannel(server_address_.str(), channel_creds, args); + !(GetParam().inproc) + ? CreateCustomChannel(server_address_.str(), channel_creds, args) + : server_->InProcessChannel(args); std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub; stub = grpc::testing::UnimplementedEchoService::NewStub(channel); EchoRequest send_request; @@ -1634,13 +1641,17 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest { // This is expected to succeed in all cases cli_stream->WritesDone(tag(7)); verif.Expect(7, true); - got_tag = verif.Next(cq_.get(), ignore_cq_result); + // TODO(vjpai): Consider whether the following is too flexible + // or whether it should just be reset to ignore_cq_result + bool ignore_cq_wd_result = + ignore_cq_result || (server_try_cancel == CANCEL_BEFORE_PROCESSING); + got_tag = verif.Next(cq_.get(), ignore_cq_wd_result); GPR_ASSERT((got_tag == 7) || (got_tag == 11 && want_done_tag)); if (got_tag == 11) { EXPECT_TRUE(srv_ctx.IsCancelled()); want_done_tag = false; // Now get the other entry that we were waiting on - EXPECT_EQ(verif.Next(cq_.get(), ignore_cq_result), 7); + EXPECT_EQ(verif.Next(cq_.get(), ignore_cq_wd_result), 7); } // This is expected to fail in all cases i.e for all values of @@ -1732,8 +1743,14 @@ std::vector<TestScenario> CreateTestScenarios(bool test_disable_blocking, std::vector<grpc::string> credentials_types; std::vector<grpc::string> messages; - if (GetCredentialsProvider()->GetChannelCredentials(kInsecureCredentialsType, - nullptr) != nullptr) { + auto insec_ok = [] { + // Only allow insecure credentials type when it is registered with the + // provider. User may create providers that do not have insecure. + return GetCredentialsProvider()->GetChannelCredentials( + kInsecureCredentialsType, nullptr) != nullptr; + }; + + if (insec_ok()) { credentials_types.push_back(kInsecureCredentialsType); } auto sec_list = GetCredentialsProvider()->GetSecureCredentialsTypeList(); @@ -1752,15 +1769,22 @@ std::vector<TestScenario> CreateTestScenarios(bool test_disable_blocking, messages.push_back(big_msg); } - for (auto health_check_service : {false, true}) { - for (auto cred = credentials_types.begin(); cred != credentials_types.end(); - ++cred) { - for (auto msg = messages.begin(); msg != messages.end(); msg++) { - scenarios.emplace_back(false, *cred, health_check_service, *msg); + // TODO (sreek) Renable tests with health check service after the issue + // https://github.com/grpc/grpc/issues/11223 is resolved + for (auto health_check_service : {false}) { + for (auto msg = messages.begin(); msg != messages.end(); msg++) { + for (auto cred = credentials_types.begin(); + cred != credentials_types.end(); ++cred) { + scenarios.emplace_back(false, false, *cred, health_check_service, *msg); if (test_disable_blocking) { - scenarios.emplace_back(true, *cred, health_check_service, *msg); + scenarios.emplace_back(true, false, *cred, health_check_service, + *msg); } } + if (insec_ok()) { + scenarios.emplace_back(false, true, kInsecureCredentialsType, + health_check_service, *msg); + } } } return scenarios; diff --git a/test/cpp/end2end/client_lb_end2end_test.cc b/test/cpp/end2end/client_lb_end2end_test.cc index f71e557450..e1160ecdc6 100644 --- a/test/cpp/end2end/client_lb_end2end_test.cc +++ b/test/cpp/end2end/client_lb_end2end_test.cc @@ -21,8 +21,6 @@ #include <mutex> #include <thread> -#include <gtest/gtest.h> - #include <grpc++/channel.h> #include <grpc++/client_context.h> #include <grpc++/create_channel.h> @@ -37,6 +35,7 @@ extern "C" { #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h" +#include "src/core/ext/filters/client_channel/subchannel_index.h" } #include "src/proto/grpc/testing/echo.grpc.pb.h" @@ -44,6 +43,8 @@ extern "C" { #include "test/core/util/test_config.h" #include "test/cpp/end2end/test_service_impl.h" +#include <gtest/gtest.h> + using grpc::testing::EchoRequest; using grpc::testing::EchoResponse; using std::chrono::system_clock; @@ -331,10 +332,14 @@ TEST_F(ClientLbEnd2endTest, PickFirstManyUpdates) { for (size_t i = 0; i < servers_.size(); ++i) { ports.emplace_back(servers_[i]->port_); } - for (size_t i = 0; i < 1000; ++i) { - std::random_shuffle(ports.begin(), ports.end()); - SetNextResolution(ports); - if (i % 10 == 0) SendRpc(); + for (const bool force_creation : {true, false}) { + grpc_subchannel_index_test_only_set_force_creation(force_creation); + gpr_log(GPR_INFO, "Force subchannel creation: %d", force_creation); + for (size_t i = 0; i < 1000; ++i) { + std::random_shuffle(ports.begin(), ports.end()); + SetNextResolution(ports); + if (i % 10 == 0) SendRpc(); + } } // Check LB policy name for the channel. EXPECT_EQ("pick_first", channel_->GetLoadBalancingPolicyName()); @@ -463,6 +468,11 @@ TEST_F(ClientLbEnd2endTest, RoundRobinManyUpdates) { EXPECT_EQ("round_robin", channel_->GetLoadBalancingPolicyName()); } +TEST_F(ClientLbEnd2endTest, RoundRobinConcurrentUpdates) { + // TODO(dgq): replicate the way internal testing exercises the concurrent + // update provisions of RR. +} + TEST_F(ClientLbEnd2endTest, RoundRobinReconnect) { // Start servers and send one RPC per server. const int kNumServers = 1; diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index da1c9b1f15..8bada48a2b 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -193,10 +193,11 @@ class TestServiceImplDupPkg class TestScenario { public: - TestScenario(bool proxy, const grpc::string& creds_type) - : use_proxy(proxy), credentials_type(creds_type) {} + TestScenario(bool proxy, bool inproc_stub, const grpc::string& creds_type) + : use_proxy(proxy), inproc(inproc_stub), credentials_type(creds_type) {} void Log() const; bool use_proxy; + bool inproc; // Although the below grpc::string is logically const, we can't declare // them const because of a limitation in the way old compilers (e.g., gcc-4.4) // manage vector insertion using a copy constructor @@ -206,8 +207,9 @@ class TestScenario { static std::ostream& operator<<(std::ostream& out, const TestScenario& scenario) { return out << "TestScenario{use_proxy=" - << (scenario.use_proxy ? "true" : "false") << ", credentials='" - << scenario.credentials_type << "'}"; + << (scenario.use_proxy ? "true" : "false") + << ", inproc=" << (scenario.inproc ? "true" : "false") + << ", credentials='" << scenario.credentials_type << "'}"; } void TestScenario::Log() const { @@ -273,7 +275,13 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> { args.SetUserAgentPrefix(user_agent_prefix_); } args.SetString(GRPC_ARG_SECONDARY_USER_AGENT_STRING, "end2end_test"); - channel_ = CreateCustomChannel(server_address_.str(), channel_creds, args); + + if (!GetParam().inproc) { + channel_ = + CreateCustomChannel(server_address_.str(), channel_creds, args); + } else { + channel_ = server_->InProcessChannel(args); + } } void ResetStub() { @@ -437,7 +445,7 @@ class End2endServerTryCancelTest : public End2endTest { auto stream = stub_->ResponseStream(&context, request); int num_msgs_read = 0; - while (num_msgs_read < kNumResponseStreamsMsgs) { + while (num_msgs_read < kServerDefaultResponseStreamsToSend) { if (!stream->Read(&response)) { break; } @@ -463,14 +471,14 @@ class End2endServerTryCancelTest : public End2endTest { case CANCEL_DURING_PROCESSING: // Server cancelled while writing messages. Client must have read less // than or equal to the expected number of messages - EXPECT_LE(num_msgs_read, kNumResponseStreamsMsgs); + EXPECT_LE(num_msgs_read, kServerDefaultResponseStreamsToSend); break; case CANCEL_AFTER_PROCESSING: // Even though the Server cancelled after writing all messages, the RPC // may be cancelled before the Client got a chance to read all the // messages. - EXPECT_LE(num_msgs_read, kNumResponseStreamsMsgs); + EXPECT_LE(num_msgs_read, kServerDefaultResponseStreamsToSend); break; default: { @@ -633,6 +641,10 @@ TEST_P(End2endServerTryCancelTest, BidiStreamServerCancelAfter) { } TEST_P(End2endTest, SimpleRpcWithCustomUserAgentPrefix) { + // User-Agent is an HTTP header for HTTP transports only + if (GetParam().inproc) { + return; + } user_agent_prefix_ = "custom_prefix"; ResetStub(); EchoRequest request; @@ -743,12 +755,10 @@ TEST_P(End2endTest, ResponseStream) { request.set_message("hello"); auto stream = stub_->ResponseStream(&context, request); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message() + "0"); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message() + "1"); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message() + "2"); + for (int i = 0; i < kServerDefaultResponseStreamsToSend; ++i) { + EXPECT_TRUE(stream->Read(&response)); + EXPECT_EQ(response.message(), request.message() + grpc::to_string(i)); + } EXPECT_FALSE(stream->Read(&response)); Status s = stream->Finish(); @@ -764,12 +774,33 @@ TEST_P(End2endTest, ResponseStreamWithCoalescingApi) { context.AddMetadata(kServerUseCoalescingApi, "1"); auto stream = stub_->ResponseStream(&context, request); + for (int i = 0; i < kServerDefaultResponseStreamsToSend; ++i) { + EXPECT_TRUE(stream->Read(&response)); + EXPECT_EQ(response.message(), request.message() + grpc::to_string(i)); + } + EXPECT_FALSE(stream->Read(&response)); + + Status s = stream->Finish(); + EXPECT_TRUE(s.ok()); +} + +// This was added to prevent regression from issue: +// https://github.com/grpc/grpc/issues/11546 +TEST_P(End2endTest, ResponseStreamWithEverythingCoalesced) { + ResetStub(); + EchoRequest request; + EchoResponse response; + ClientContext context; + request.set_message("hello"); + context.AddMetadata(kServerUseCoalescingApi, "1"); + // We will only send one message, forcing everything (init metadata, message, + // trailing) to be coalesced together. + context.AddMetadata(kServerResponseStreamsToSend, "1"); + + auto stream = stub_->ResponseStream(&context, request); EXPECT_TRUE(stream->Read(&response)); EXPECT_EQ(response.message(), request.message() + "0"); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message() + "1"); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message() + "2"); + EXPECT_FALSE(stream->Read(&response)); Status s = stream->Finish(); @@ -785,20 +816,12 @@ TEST_P(End2endTest, BidiStream) { auto stream = stub_->BidiStream(&context); - request.set_message(msg + "0"); - EXPECT_TRUE(stream->Write(request)); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message()); - - request.set_message(msg + "1"); - EXPECT_TRUE(stream->Write(request)); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message()); - - request.set_message(msg + "2"); - EXPECT_TRUE(stream->Write(request)); - EXPECT_TRUE(stream->Read(&response)); - EXPECT_EQ(response.message(), request.message()); + for (int i = 0; i < kServerDefaultResponseStreamsToSend; ++i) { + request.set_message(msg + grpc::to_string(i)); + EXPECT_TRUE(stream->Write(request)); + EXPECT_TRUE(stream->Read(&response)); + EXPECT_EQ(response.message(), request.message()); + } stream->WritesDone(); EXPECT_FALSE(stream->Read(&response)); @@ -841,6 +864,31 @@ TEST_P(End2endTest, BidiStreamWithCoalescingApi) { EXPECT_TRUE(s.ok()); } +// This was added to prevent regression from issue: +// https://github.com/grpc/grpc/issues/11546 +TEST_P(End2endTest, BidiStreamWithEverythingCoalesced) { + ResetStub(); + EchoRequest request; + EchoResponse response; + ClientContext context; + context.AddMetadata(kServerFinishAfterNReads, "1"); + context.set_initial_metadata_corked(true); + grpc::string msg("hello"); + + auto stream = stub_->BidiStream(&context); + + request.set_message(msg + "0"); + stream->WriteLast(request, WriteOptions()); + EXPECT_TRUE(stream->Read(&response)); + EXPECT_EQ(response.message(), request.message()); + + EXPECT_FALSE(stream->Read(&response)); + EXPECT_FALSE(stream->Read(&response)); + + Status s = stream->Finish(); + EXPECT_TRUE(s.ok()); +} + // Talk to the two services with the same name but different package names. // The two stubs are created on the same channel. TEST_P(End2endTest, DiffPackageServices) { @@ -1029,6 +1077,10 @@ TEST_P(End2endTest, SimultaneousReadWritesDone) { } TEST_P(End2endTest, ChannelState) { + if (GetParam().inproc) { + return; + } + ResetStub(); // Start IDLE EXPECT_EQ(GRPC_CHANNEL_IDLE, channel_->GetState(false)); @@ -1052,7 +1104,8 @@ TEST_P(End2endTest, ChannelState) { // Takes 10s. TEST_P(End2endTest, ChannelStateTimeout) { - if (GetParam().credentials_type != kInsecureCredentialsType) { + if ((GetParam().credentials_type != kInsecureCredentialsType) || + GetParam().inproc) { return; } int port = grpc_pick_unused_port_or_die(); @@ -1512,7 +1565,9 @@ TEST_P(SecureEnd2endTest, NonBlockingAuthMetadataPluginFailure) { Status s = stub_->Echo(&context, request, &response); EXPECT_FALSE(s.ok()); EXPECT_EQ(s.error_code(), StatusCode::UNAUTHENTICATED); - EXPECT_EQ(s.error_message(), kTestCredsPluginErrorMsg); + EXPECT_EQ(s.error_message(), + grpc::string("Getting metadata from plugin failed with error: ") + + kTestCredsPluginErrorMsg); } TEST_P(SecureEnd2endTest, NonBlockingAuthMetadataPluginAndProcessorSuccess) { @@ -1571,7 +1626,9 @@ TEST_P(SecureEnd2endTest, BlockingAuthMetadataPluginFailure) { Status s = stub_->Echo(&context, request, &response); EXPECT_FALSE(s.ok()); EXPECT_EQ(s.error_code(), StatusCode::UNAUTHENTICATED); - EXPECT_EQ(s.error_message(), kTestCredsPluginErrorMsg); + EXPECT_EQ(s.error_message(), + grpc::string("Getting metadata from plugin failed with error: ") + + kTestCredsPluginErrorMsg); } TEST_P(SecureEnd2endTest, ClientAuthContext) { @@ -1633,51 +1690,56 @@ TEST_P(ResourceQuotaEnd2endTest, SimpleRequest) { std::vector<TestScenario> CreateTestScenarios(bool use_proxy, bool test_insecure, - bool test_secure) { + bool test_secure, + bool test_inproc) { std::vector<TestScenario> scenarios; std::vector<grpc::string> credentials_types; if (test_secure) { credentials_types = GetCredentialsProvider()->GetSecureCredentialsTypeList(); } - if (test_insecure) { - // Only add insecure credentials type when it is registered with the + auto insec_ok = [] { + // Only allow insecure credentials type when it is registered with the // provider. User may create providers that do not have insecure. - if (GetCredentialsProvider()->GetChannelCredentials( - kInsecureCredentialsType, nullptr) != nullptr) { - credentials_types.push_back(kInsecureCredentialsType); - } + return GetCredentialsProvider()->GetChannelCredentials( + kInsecureCredentialsType, nullptr) != nullptr; + }; + if (test_insecure && insec_ok()) { + credentials_types.push_back(kInsecureCredentialsType); } GPR_ASSERT(!credentials_types.empty()); for (auto it = credentials_types.begin(); it != credentials_types.end(); ++it) { - scenarios.emplace_back(false, *it); + scenarios.emplace_back(false, false, *it); if (use_proxy) { - scenarios.emplace_back(true, *it); + scenarios.emplace_back(true, false, *it); } } + if (test_inproc && insec_ok()) { + scenarios.emplace_back(false, true, kInsecureCredentialsType); + } return scenarios; } INSTANTIATE_TEST_CASE_P(End2end, End2endTest, ::testing::ValuesIn(CreateTestScenarios(false, true, - true))); + true, true))); INSTANTIATE_TEST_CASE_P(End2endServerTryCancel, End2endServerTryCancelTest, ::testing::ValuesIn(CreateTestScenarios(false, true, - true))); + true, true))); INSTANTIATE_TEST_CASE_P(ProxyEnd2end, ProxyEnd2endTest, ::testing::ValuesIn(CreateTestScenarios(true, true, - true))); + true, false))); INSTANTIATE_TEST_CASE_P(SecureEnd2end, SecureEnd2endTest, ::testing::ValuesIn(CreateTestScenarios(false, false, - true))); + true, false))); INSTANTIATE_TEST_CASE_P(ResourceQuotaEnd2end, ResourceQuotaEnd2endTest, ::testing::ValuesIn(CreateTestScenarios(false, true, - true))); + true, true))); } // namespace } // namespace testing diff --git a/test/cpp/end2end/grpclb_end2end_test.cc b/test/cpp/end2end/grpclb_end2end_test.cc index 4f839f7788..86cce2d30d 100644 --- a/test/cpp/end2end/grpclb_end2end_test.cc +++ b/test/cpp/end2end/grpclb_end2end_test.cc @@ -32,7 +32,6 @@ #include <grpc/support/string_util.h> #include <grpc/support/thd.h> #include <grpc/support/time.h> -#include <gtest/gtest.h> extern "C" { #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h" @@ -46,6 +45,8 @@ extern "C" { #include "src/proto/grpc/lb/v1/load_balancer.grpc.pb.h" #include "src/proto/grpc/testing/echo.grpc.pb.h" +#include <gtest/gtest.h> + // TODO(dgq): Other scenarios in need of testing: // - Send a serverlist with faulty ip:port addresses (port > 2^16, etc). // - Test reception of invalid serverlist @@ -194,12 +195,13 @@ class BalancerServiceImpl : public BalancerService { for (const auto& response_and_delay : responses_and_delays) { { std::unique_lock<std::mutex> lock(mu_); - if (shutdown_) break; + if (shutdown_) goto done; } SendResponse(stream, response_and_delay.first, response_and_delay.second); } { std::unique_lock<std::mutex> lock(mu_); + if (shutdown_) goto done; serverlist_cond_.wait(lock); } @@ -209,6 +211,9 @@ class BalancerServiceImpl : public BalancerService { gpr_log(GPR_INFO, "LB: recv client load report msg: '%s'", request.DebugString().c_str()); GPR_ASSERT(request.has_client_stats()); + // We need to acquire the lock here in order to prevent the notify_one + // below from firing before its corresponding wait is executed. + std::lock_guard<std::mutex> lock(mu_); client_stats_.num_calls_started += request.client_stats().num_calls_started(); client_stats_.num_calls_finished += @@ -224,10 +229,9 @@ class BalancerServiceImpl : public BalancerService { .num_calls_finished_with_client_failed_to_send(); client_stats_.num_calls_finished_known_received += request.client_stats().num_calls_finished_known_received(); - std::lock_guard<std::mutex> lock(mu_); load_report_cond_.notify_one(); } - + done: gpr_log(GPR_INFO, "LB: done"); return Status::OK; } @@ -428,19 +432,24 @@ class GrpclbEnd2endTest : public ::testing::Test { explicit ServerThread(const grpc::string& type, const grpc::string& server_host, T* service) : type_(type), service_(service) { + std::mutex mu; + // We need to acquire the lock here in order to prevent the notify_one + // by ServerThread::Start from firing before the wait below is hit. + std::unique_lock<std::mutex> lock(mu); port_ = grpc_pick_unused_port_or_die(); gpr_log(GPR_INFO, "starting %s server on port %d", type_.c_str(), port_); - std::mutex mu; std::condition_variable cond; thread_.reset(new std::thread( std::bind(&ServerThread::Start, this, server_host, &mu, &cond))); - std::unique_lock<std::mutex> lock(mu); cond.wait(lock); gpr_log(GPR_INFO, "%s server startup complete", type_.c_str()); } void Start(const grpc::string& server_host, std::mutex* mu, std::condition_variable* cond) { + // We need to acquire the lock here in order to prevent the notify_one + // below from firing before its corresponding wait is executed. + std::lock_guard<std::mutex> lock(*mu); std::ostringstream server_address; server_address << server_host << ":" << port_; ServerBuilder builder; @@ -448,13 +457,12 @@ class GrpclbEnd2endTest : public ::testing::Test { InsecureServerCredentials()); builder.RegisterService(service_); server_ = builder.BuildAndStart(); - std::lock_guard<std::mutex> lock(*mu); cond->notify_one(); } void Shutdown() { gpr_log(GPR_INFO, "%s about to shutdown", type_.c_str()); - server_->Shutdown(); + server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0)); thread_->join(); gpr_log(GPR_INFO, "%s shutdown completed", type_.c_str()); } @@ -820,6 +828,7 @@ TEST_F(UpdatesTest, UpdateBalancersDeadUpdate) { // Kill balancer 0 gpr_log(GPR_INFO, "********** ABOUT TO KILL BALANCER 0 *************"); + balancers_[0]->NotifyDoneWithServerlists(); if (balancers_[0]->Shutdown()) balancer_servers_[0].Shutdown(); gpr_log(GPR_INFO, "********** KILLED BALANCER 0 *************"); diff --git a/test/cpp/end2end/hybrid_end2end_test.cc b/test/cpp/end2end/hybrid_end2end_test.cc index 8a31ab77b2..cb515533ed 100644 --- a/test/cpp/end2end/hybrid_end2end_test.cc +++ b/test/cpp/end2end/hybrid_end2end_test.cc @@ -521,7 +521,7 @@ class SplitResponseStreamDupPkg stream->NextMessageSize(&next_msg_sz); gpr_log(GPR_INFO, "Split Streamed Next Message Size is %u", next_msg_sz); GPR_ASSERT(stream->Read(&req)); - for (int i = 0; i < kNumResponseStreamsMsgs; i++) { + for (int i = 0; i < kServerDefaultResponseStreamsToSend; i++) { resp.set_message(req.message() + grpc::to_string(i) + "_dup"); GPR_ASSERT(stream->Write(resp)); } @@ -561,7 +561,7 @@ class FullySplitStreamedDupPkg stream->NextMessageSize(&next_msg_sz); gpr_log(GPR_INFO, "Split Streamed Next Message Size is %u", next_msg_sz); GPR_ASSERT(stream->Read(&req)); - for (int i = 0; i < kNumResponseStreamsMsgs; i++) { + for (int i = 0; i < kServerDefaultResponseStreamsToSend; i++) { resp.set_message(req.message() + grpc::to_string(i) + "_dup"); GPR_ASSERT(stream->Write(resp)); } @@ -613,7 +613,7 @@ class FullyStreamedDupPkg : public duplicate::EchoTestService::StreamedService { stream->NextMessageSize(&next_msg_sz); gpr_log(GPR_INFO, "Split Streamed Next Message Size is %u", next_msg_sz); GPR_ASSERT(stream->Read(&req)); - for (int i = 0; i < kNumResponseStreamsMsgs; i++) { + for (int i = 0; i < kServerDefaultResponseStreamsToSend; i++) { resp.set_message(req.message() + grpc::to_string(i) + "_dup"); GPR_ASSERT(stream->Write(resp)); } diff --git a/test/cpp/end2end/mock_test.cc b/test/cpp/end2end/mock_test.cc index 8ca91363d8..0b63c25055 100644 --- a/test/cpp/end2end/mock_test.cc +++ b/test/cpp/end2end/mock_test.cc @@ -19,7 +19,6 @@ #include <climits> #include <thread> -#include <gmock/gmock.h> #include <grpc++/channel.h> #include <grpc++/client_context.h> #include <grpc++/create_channel.h> @@ -31,14 +30,15 @@ #include <grpc/support/thd.h> #include <grpc/support/time.h> -#include <grpc++/test/mock_stream.h> - #include "src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.h" #include "src/proto/grpc/testing/echo.grpc.pb.h" #include "src/proto/grpc/testing/echo_mock.grpc.pb.h" #include "test/core/util/port.h" #include "test/core/util/test_config.h" +#include <grpc++/test/mock_stream.h> + +#include <gmock/gmock.h> #include <gtest/gtest.h> #include <iostream> diff --git a/test/cpp/end2end/round_robin_end2end_test.cc b/test/cpp/end2end/round_robin_end2end_test.cc index 0cb727f0ed..eee32ce85d 100644 --- a/test/cpp/end2end/round_robin_end2end_test.cc +++ b/test/cpp/end2end/round_robin_end2end_test.cc @@ -73,9 +73,12 @@ class RoundRobinEnd2endTest : public ::testing::Test { protected: RoundRobinEnd2endTest() : server_host_("localhost") {} - void StartServers(int num_servers) { - for (int i = 0; i < num_servers; ++i) { - servers_.emplace_back(new ServerData(server_host_)); + void StartServers(size_t num_servers, + std::vector<int> ports = std::vector<int>()) { + for (size_t i = 0; i < num_servers; ++i) { + int port = 0; + if (ports.size() == num_servers) port = ports[i]; + servers_.emplace_back(new ServerData(server_host_, port)); } } @@ -99,15 +102,19 @@ class RoundRobinEnd2endTest : public ::testing::Test { stub_ = grpc::testing::EchoTestService::NewStub(channel_); } - void SendRpc(int num_rpcs) { + void SendRpc(int num_rpcs, bool expect_ok = true) { EchoRequest request; EchoResponse response; request.set_message("Live long and prosper."); for (int i = 0; i < num_rpcs; i++) { ClientContext context; Status status = stub_->Echo(&context, request, &response); - EXPECT_TRUE(status.ok()); - EXPECT_EQ(response.message(), request.message()); + if (expect_ok) { + EXPECT_TRUE(status.ok()); + EXPECT_EQ(response.message(), request.message()); + } else { + EXPECT_FALSE(status.ok()); + } } } @@ -116,8 +123,8 @@ class RoundRobinEnd2endTest : public ::testing::Test { std::unique_ptr<Server> server_; MyTestServiceImpl service_; - explicit ServerData(const grpc::string& server_host) { - port_ = grpc_pick_unused_port_or_die(); + explicit ServerData(const grpc::string& server_host, int port = 0) { + port_ = port > 0 ? port : grpc_pick_unused_port_or_die(); gpr_log(GPR_INFO, "starting server on port %d", port_); std::ostringstream server_address; server_address << server_host << ":" << port_; @@ -176,6 +183,38 @@ TEST_F(RoundRobinEnd2endTest, RoundRobin) { EXPECT_EQ("round_robin", channel_->GetLoadBalancingPolicyName()); } +TEST_F(RoundRobinEnd2endTest, RoundRobinReconnect) { + // Start servers and send one RPC per server. + const int kNumServers = 1; + std::vector<int> ports; + ports.push_back(grpc_pick_unused_port_or_die()); + StartServers(kNumServers, ports); + ResetStub(true /* round_robin */); + // Send one RPC per backend and make sure they are used in order. + // Note: This relies on the fact that the subchannels are reported in + // state READY in the order in which the addresses are specified, + // which is only true because the backends are all local. + for (size_t i = 0; i < servers_.size(); ++i) { + SendRpc(1); + EXPECT_EQ(1, servers_[i]->service_.request_count()) << "for backend #" << i; + } + // Check LB policy name for the channel. + EXPECT_EQ("round_robin", channel_->GetLoadBalancingPolicyName()); + + // Kill all servers + for (size_t i = 0; i < servers_.size(); ++i) { + servers_[i]->Shutdown(); + } + // Client request should fail. + SendRpc(1, false); + + // Bring servers back up on the same port (we aren't recreating the channel). + StartServers(kNumServers, ports); + + // Client request should succeed. + SendRpc(1); +} + } // namespace } // namespace testing } // namespace grpc diff --git a/test/cpp/end2end/test_service_impl.cc b/test/cpp/end2end/test_service_impl.cc index e1260277b4..4fa98c24f5 100644 --- a/test/cpp/end2end/test_service_impl.cc +++ b/test/cpp/end2end/test_service_impl.cc @@ -239,6 +239,10 @@ Status TestServiceImpl::ResponseStream(ServerContext* context, int server_coalescing_api = GetIntValueFromMetadata( kServerUseCoalescingApi, context->client_metadata(), 0); + int server_responses_to_send = GetIntValueFromMetadata( + kServerResponseStreamsToSend, context->client_metadata(), + kServerDefaultResponseStreamsToSend); + if (server_try_cancel == CANCEL_BEFORE_PROCESSING) { ServerTryCancel(context); return Status::CANCELLED; @@ -251,9 +255,9 @@ Status TestServiceImpl::ResponseStream(ServerContext* context, new std::thread(&TestServiceImpl::ServerTryCancel, this, context); } - for (int i = 0; i < kNumResponseStreamsMsgs; i++) { + for (int i = 0; i < server_responses_to_send; i++) { response.set_message(request->message() + grpc::to_string(i)); - if (i == kNumResponseStreamsMsgs - 1 && server_coalescing_api != 0) { + if (i == server_responses_to_send - 1 && server_coalescing_api != 0) { writer->WriteLast(response, WriteOptions()); } else { writer->Write(response); diff --git a/test/cpp/end2end/test_service_impl.h b/test/cpp/end2end/test_service_impl.h index 52f1b991c7..e485769bb2 100644 --- a/test/cpp/end2end/test_service_impl.h +++ b/test/cpp/end2end/test_service_impl.h @@ -29,7 +29,8 @@ namespace grpc { namespace testing { -const int kNumResponseStreamsMsgs = 3; +const int kServerDefaultResponseStreamsToSend = 3; +const char* const kServerResponseStreamsToSend = "server_responses_to_send"; const char* const kServerCancelAfterReads = "cancel_after_reads"; const char* const kServerTryCancelRequest = "server_try_cancel"; const char* const kDebugInfoTrailerKey = "debug-info-bin"; diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc index 542df00fb4..f990a7ed9d 100644 --- a/test/cpp/end2end/thread_stress_test.cc +++ b/test/cpp/end2end/thread_stress_test.cc @@ -151,16 +151,6 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service { std::mutex mu_; }; -class TestServiceImplDupPkg - : public ::grpc::testing::duplicate::EchoTestService::Service { - public: - Status Echo(ServerContext* context, const EchoRequest* request, - EchoResponse* response) override { - response->set_message("no package"); - return Status::OK; - } -}; - template <class Service> class CommonStressTest { public: @@ -168,63 +158,92 @@ class CommonStressTest { virtual ~CommonStressTest() {} virtual void SetUp() = 0; virtual void TearDown() = 0; - void ResetStub() { - std::shared_ptr<Channel> channel = - CreateChannel(server_address_.str(), InsecureChannelCredentials()); - stub_ = grpc::testing::EchoTestService::NewStub(channel); - } + virtual void ResetStub() = 0; grpc::testing::EchoTestService::Stub* GetStub() { return stub_.get(); } protected: - void SetUpStart(ServerBuilder* builder, Service* service) { - int port = grpc_pick_unused_port_or_die(); - server_address_ << "localhost:" << port; - // Setup server - builder->AddListeningPort(server_address_.str(), - InsecureServerCredentials()); + std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_; + std::unique_ptr<Server> server_; + + virtual void SetUpStart(ServerBuilder* builder, Service* service) = 0; + void SetUpStartCommon(ServerBuilder* builder, Service* service) { builder->RegisterService(service); builder->SetMaxMessageSize( kMaxMessageSize_); // For testing max message size. - builder->RegisterService(&dup_pkg_service_); } void SetUpEnd(ServerBuilder* builder) { server_ = builder->BuildAndStart(); } void TearDownStart() { server_->Shutdown(); } void TearDownEnd() {} private: - std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_; - std::unique_ptr<Server> server_; - std::ostringstream server_address_; const int kMaxMessageSize_; - TestServiceImplDupPkg dup_pkg_service_; }; -class CommonStressTestSyncServer : public CommonStressTest<TestServiceImpl> { +template <class Service> +class CommonStressTestInsecure : public CommonStressTest<Service> { + public: + void ResetStub() override { + std::shared_ptr<Channel> channel = + CreateChannel(server_address_.str(), InsecureChannelCredentials()); + this->stub_ = grpc::testing::EchoTestService::NewStub(channel); + } + + protected: + void SetUpStart(ServerBuilder* builder, Service* service) override { + int port = grpc_pick_unused_port_or_die(); + this->server_address_ << "localhost:" << port; + // Setup server + builder->AddListeningPort(server_address_.str(), + InsecureServerCredentials()); + this->SetUpStartCommon(builder, service); + } + + private: + std::ostringstream server_address_; +}; + +template <class Service> +class CommonStressTestInproc : public CommonStressTest<Service> { + public: + void ResetStub() override { + ChannelArguments args; + std::shared_ptr<Channel> channel = this->server_->InProcessChannel(args); + this->stub_ = grpc::testing::EchoTestService::NewStub(channel); + } + + protected: + void SetUpStart(ServerBuilder* builder, Service* service) override { + this->SetUpStartCommon(builder, service); + } +}; + +template <class BaseClass> +class CommonStressTestSyncServer : public BaseClass { public: void SetUp() override { ServerBuilder builder; - SetUpStart(&builder, &service_); - SetUpEnd(&builder); + this->SetUpStart(&builder, &service_); + this->SetUpEnd(&builder); } void TearDown() override { - TearDownStart(); - TearDownEnd(); + this->TearDownStart(); + this->TearDownEnd(); } private: TestServiceImpl service_; }; -class CommonStressTestAsyncServer - : public CommonStressTest<grpc::testing::EchoTestService::AsyncService> { +template <class BaseClass> +class CommonStressTestAsyncServer : public BaseClass { public: CommonStressTestAsyncServer() : contexts_(kNumAsyncServerThreads * 100) {} void SetUp() override { shutting_down_ = false; ServerBuilder builder; - SetUpStart(&builder, &service_); + this->SetUpStart(&builder, &service_); cq_ = builder.AddCompletionQueue(); - SetUpEnd(&builder); + this->SetUpEnd(&builder); for (int i = 0; i < kNumAsyncServerThreads * 100; i++) { RefreshContext(i); } @@ -236,7 +255,7 @@ class CommonStressTestAsyncServer void TearDown() override { { std::unique_lock<std::mutex> l(mu_); - TearDownStart(); + this->TearDownStart(); shutting_down_ = true; cq_->Shutdown(); } @@ -249,7 +268,7 @@ class CommonStressTestAsyncServer bool ignored_ok; while (cq_->Next(&ignored_tag, &ignored_ok)) ; - TearDownEnd(); + this->TearDownEnd(); } private: @@ -332,8 +351,13 @@ static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs) { } } -typedef ::testing::Types<CommonStressTestSyncServer, - CommonStressTestAsyncServer> +typedef ::testing::Types< + CommonStressTestSyncServer<CommonStressTestInsecure<TestServiceImpl>>, + CommonStressTestSyncServer<CommonStressTestInproc<TestServiceImpl>>, + CommonStressTestAsyncServer< + CommonStressTestInsecure<grpc::testing::EchoTestService::AsyncService>>, + CommonStressTestAsyncServer< + CommonStressTestInproc<grpc::testing::EchoTestService::AsyncService>>> CommonTypes; TYPED_TEST_CASE(End2endTest, CommonTypes); TYPED_TEST(End2endTest, ThreadStress) { diff --git a/test/cpp/interop/BUILD b/test/cpp/interop/BUILD index 93e01fee4b..9123bd929e 100644 --- a/test/cpp/interop/BUILD +++ b/test/cpp/interop/BUILD @@ -14,7 +14,16 @@ licenses(["notice"]) # Apache v2 -cc_library( +load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_test", "grpc_cc_binary") + +package( + features = [ + "-layering_check", + "-parse_headers", + ], +) + +grpc_cc_library( name = "server_helper_lib", srcs = [ "server_helper.cc", @@ -22,21 +31,35 @@ cc_library( hdrs = [ "server_helper.h", ], + external_deps = [ + "gflags", + ], + language = "C++", deps = [ "//test/cpp/util:test_util", - "//external:gflags", ], ) -cc_binary( +grpc_cc_binary( name = "interop_server", srcs = [ - "interop_server.cc", "interop_server_bootstrap.cc", ], + language = "C++", deps = [ - ":server_helper_lib", + ":interop_server_lib", "//:grpc++", + ], +) + +grpc_cc_library( + name = "interop_server_lib", + srcs = [ + "interop_server.cc", + ], + language = "C++", + deps = [ + ":server_helper_lib", "//src/proto/grpc/testing:empty_proto", "//src/proto/grpc/testing:messages_proto", "//src/proto/grpc/testing:test_proto", @@ -44,7 +67,7 @@ cc_binary( ], ) -cc_library( +grpc_cc_library( name = "client_helper_lib", srcs = [ "client_helper.cc", @@ -54,22 +77,73 @@ cc_library( "client_helper.h", "interop_client.h", ], + language = "C++", deps = [ - "//test/cpp/util:test_util", "//src/proto/grpc/testing:empty_proto", "//src/proto/grpc/testing:messages_proto", "//src/proto/grpc/testing:test_proto", "//test/core/security:oauth2_utils", "//test/cpp/util:test_config", + "//test/cpp/util:test_util", ], ) -cc_binary( - name = "interop_client", +grpc_cc_library( + name = "interop_client_main", srcs = [ "client.cc", ], + language = "C++", + deps = [ + ":client_helper_lib", + ], +) + +grpc_cc_binary( + name = "interop_client", + language = "C++", + deps = [ + ":interop_client_main", + "//:grpc++", + ], +) + +grpc_cc_binary( + name = "reconnect_interop_client", + srcs = [ + "reconnect_interop_client.cc", + ], + deps = [ + ":client_helper_lib", + "//:grpc++", + "//test/core/util:grpc_test_util", + ], +) + +grpc_cc_binary( + name = "reconnect_interop_server", + srcs = [ + "reconnect_interop_server.cc", + ], + language = "C++", + deps = [ + ":interop_server_lib", + "//:grpc++", + "//test/core/util:grpc_test_util", + ], +) + +grpc_cc_binary( + name = "stress_test", + srcs = [ + "stress_interop_client.cc", + "stress_interop_client.h", + "stress_test.cc", + ], deps = [ ":client_helper_lib", + "//:grpc++", + "//test/cpp/util:metrics_server_lib", + "//test/cpp/util:test_config", ], ) diff --git a/test/cpp/microbenchmarks/BUILD b/test/cpp/microbenchmarks/BUILD index 10df1cb781..5e1bcee89f 100644 --- a/test/cpp/microbenchmarks/BUILD +++ b/test/cpp/microbenchmarks/BUILD @@ -16,6 +16,13 @@ licenses(["notice"]) # Apache v2 load("//bazel:grpc_build_system.bzl", "grpc_cc_test", "grpc_cc_library") +package( + features = [ + "-layering_check", + "-parse_headers", + ], +) + grpc_cc_test( name = "noop-benchmark", srcs = ["noop-benchmark.cc"], diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc index 1e3830a556..f79db15a47 100644 --- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc +++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc @@ -41,6 +41,7 @@ namespace testing { static void* g_tag = (void*)(intptr_t)10; // Some random number static grpc_completion_queue* g_cq; static grpc_event_engine_vtable g_vtable; +static const grpc_event_engine_vtable* g_old_vtable; static void pollset_shutdown(grpc_exec_ctx* exec_ctx, grpc_pollset* ps, grpc_closure* closure) { @@ -72,7 +73,7 @@ static grpc_error* pollset_work(grpc_exec_ctx* exec_ctx, grpc_pollset* ps, grpc_pollset_worker** worker, gpr_timespec now, gpr_timespec deadline) { if (gpr_time_cmp(deadline, gpr_time_0(GPR_CLOCK_MONOTONIC)) == 0) { - gpr_log(GPR_ERROR, "no-op"); + gpr_log(GPR_DEBUG, "no-op"); return GRPC_ERROR_NONE; } @@ -98,7 +99,12 @@ static void init_engine_vtable() { static void setup() { grpc_init(); + + /* Override the event engine with our test event engine (g_vtable); but before + * that, save the current event engine in g_old_vtable. We will have to set + * g_old_vtable back before calling grpc_shutdown() */ init_engine_vtable(); + g_old_vtable = grpc_get_event_engine_test_only(); grpc_set_event_engine_test_only(&g_vtable); g_cq = grpc_completion_queue_create_for_next(NULL); @@ -115,6 +121,10 @@ static void teardown() { } grpc_completion_queue_destroy(g_cq); + + /* Restore the old event engine before calling grpc_shutdown */ + grpc_set_event_engine_test_only(g_old_vtable); + grpc_shutdown(); } /* A few notes about Multi-threaded benchmarks: diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc index f420bd7421..0712a40018 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc @@ -414,24 +414,34 @@ BENCHMARK_TEMPLATE(BM_StreamingPingPong, InProcessCHTTP2, NoOpMutator, ->Apply(StreamingPingPongArgs); BENCHMARK_TEMPLATE(BM_StreamingPingPong, TCP, NoOpMutator, NoOpMutator) ->Apply(StreamingPingPongArgs); +BENCHMARK_TEMPLATE(BM_StreamingPingPong, InProcess, NoOpMutator, NoOpMutator) + ->Apply(StreamingPingPongArgs); BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcessCHTTP2, NoOpMutator, NoOpMutator) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, TCP, NoOpMutator, NoOpMutator) ->Range(0, 128 * 1024 * 1024); +BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcess, NoOpMutator, + NoOpMutator) + ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_StreamingPingPong, MinInProcessCHTTP2, NoOpMutator, NoOpMutator) ->Apply(StreamingPingPongArgs); BENCHMARK_TEMPLATE(BM_StreamingPingPong, MinTCP, NoOpMutator, NoOpMutator) ->Apply(StreamingPingPongArgs); +BENCHMARK_TEMPLATE(BM_StreamingPingPong, MinInProcess, NoOpMutator, NoOpMutator) + ->Apply(StreamingPingPongArgs); BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, MinInProcessCHTTP2, NoOpMutator, NoOpMutator) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, MinTCP, NoOpMutator, NoOpMutator) ->Range(0, 128 * 1024 * 1024); +BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, MinInProcess, NoOpMutator, + NoOpMutator) + ->Range(0, 128 * 1024 * 1024); // Generate Args for StreamingPingPongWithCoalescingApi benchmarks. Currently // generates args for only "small streams" (i.e streams with 0, 1 or 2 messages) @@ -459,6 +469,12 @@ BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, InProcessCHTTP2, BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, MinInProcessCHTTP2, NoOpMutator, NoOpMutator) ->Apply(StreamingPingPongWithCoalescingApiArgs); +BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, InProcess, + NoOpMutator, NoOpMutator) + ->Apply(StreamingPingPongWithCoalescingApiArgs); +BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, MinInProcess, + NoOpMutator, NoOpMutator) + ->Apply(StreamingPingPongWithCoalescingApiArgs); } // namespace testing } // namespace grpc diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc index fc2d67ff11..6fbf9da0ad 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc @@ -173,6 +173,8 @@ BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, TCP) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, UDS) ->Range(0, 128 * 1024 * 1024); +BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, InProcess) + ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, SockPair) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, InProcessCHTTP2) @@ -181,16 +183,20 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, TCP) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, UDS) ->Range(0, 128 * 1024 * 1024); +BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcess) + ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2) ->Range(0, 128 * 1024 * 1024); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinTCP)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinUDS)->Arg(0); +BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinInProcess)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinSockPair)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinInProcessCHTTP2)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinTCP)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinUDS)->Arg(0); +BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinInProcess)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinSockPair)->Arg(0); BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinInProcessCHTTP2)->Arg(0); diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc index 6100906a9a..aeec7d831b 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc @@ -315,7 +315,7 @@ BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(StreamingTrickleArgs); static void BM_PumpUnbalancedUnary_Trickle(benchmark::State& state) { EchoTestService::AsyncService service; std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2( - &service, true, state.range(0) /* req_size */, + &service, false, state.range(0) /* req_size */, state.range(1) /* resp_size */, state.range(2) /* bw in kbit/s */)); EchoRequest send_request; EchoResponse send_response; diff --git a/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc b/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc index ee2d5ec7f4..9af751245f 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc @@ -132,6 +132,10 @@ BENCHMARK_TEMPLATE(BM_UnaryPingPong, UDS, NoOpMutator, NoOpMutator) ->Args({0, 0}); BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinUDS, NoOpMutator, NoOpMutator) ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, NoOpMutator) + ->Apply(SweepSizesArgs); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinInProcess, NoOpMutator, NoOpMutator) + ->Apply(SweepSizesArgs); BENCHMARK_TEMPLATE(BM_UnaryPingPong, SockPair, NoOpMutator, NoOpMutator) ->Args({0, 0}); BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinSockPair, NoOpMutator, NoOpMutator) @@ -191,6 +195,56 @@ BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator, BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator, Server_AddInitialMetadata<RandomAsciiMetadata<10>, 100>) ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomBinaryMetadata<10>, 1>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomBinaryMetadata<31>, 1>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomBinaryMetadata<100>, 1>, + NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomBinaryMetadata<10>, 2>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomBinaryMetadata<31>, 2>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomBinaryMetadata<100>, 2>, + NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomBinaryMetadata<10>, 1>) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomBinaryMetadata<31>, 1>) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomBinaryMetadata<100>, 1>) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomAsciiMetadata<10>, 1>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomAsciiMetadata<31>, 1>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, + Client_AddMetadata<RandomAsciiMetadata<100>, 1>, NoOpMutator) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomAsciiMetadata<10>, 1>) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomAsciiMetadata<31>, 1>) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomAsciiMetadata<100>, 1>) + ->Args({0, 0}); +BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcess, NoOpMutator, + Server_AddInitialMetadata<RandomAsciiMetadata<10>, 100>) + ->Args({0, 0}); } // namespace testing } // namespace grpc diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc index 683f4703c2..1fc1f2f83b 100644 --- a/test/cpp/microbenchmarks/bm_pollset.cc +++ b/test/cpp/microbenchmarks/bm_pollset.cc @@ -149,7 +149,7 @@ static void BM_PollAddFd(benchmark::State& state) { grpc_pollset_add_fd(&exec_ctx, ps, fd); grpc_exec_ctx_flush(&exec_ctx); } - grpc_fd_orphan(&exec_ctx, fd, NULL, NULL, "xxx"); + grpc_fd_orphan(&exec_ctx, fd, NULL, NULL, false /* already_closed */, "xxx"); grpc_closure shutdown_ps_closure; GRPC_CLOSURE_INIT(&shutdown_ps_closure, shutdown_ps, ps, grpc_schedule_on_exec_ctx); @@ -247,7 +247,8 @@ static void BM_SingleThreadPollOneFd(benchmark::State& state) { while (!done) { GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline)); } - grpc_fd_orphan(&exec_ctx, wakeup, NULL, NULL, "done"); + grpc_fd_orphan(&exec_ctx, wakeup, NULL, NULL, false /* already_closed */, + "done"); wakeup_fd.read_fd = 0; grpc_closure shutdown_ps_closure; GRPC_CLOSURE_INIT(&shutdown_ps_closure, shutdown_ps, ps, diff --git a/test/cpp/microbenchmarks/fullstack_fixtures.h b/test/cpp/microbenchmarks/fullstack_fixtures.h index 2320086afc..5477b860b4 100644 --- a/test/cpp/microbenchmarks/fullstack_fixtures.h +++ b/test/cpp/microbenchmarks/fullstack_fixtures.h @@ -66,14 +66,21 @@ class FullstackFixture : public BaseFixture { FullstackFixture(Service* service, const FixtureConfiguration& config, const grpc::string& address) { ServerBuilder b; - b.AddListeningPort(address, InsecureServerCredentials()); + if (address.length() > 0) { + b.AddListeningPort(address, InsecureServerCredentials()); + } cq_ = b.AddCompletionQueue(true); b.RegisterService(service); config.ApplyCommonServerBuilderConfig(&b); server_ = b.BuildAndStart(); ChannelArguments args; config.ApplyCommonChannelArguments(&args); - channel_ = CreateCustomChannel(address, InsecureChannelCredentials(), args); + if (address.length() > 0) { + channel_ = + CreateCustomChannel(address, InsecureChannelCredentials(), args); + } else { + channel_ = server_->InProcessChannel(args); + } } virtual ~FullstackFixture() { @@ -139,6 +146,15 @@ class UDS : public FullstackFixture { } }; +class InProcess : public FullstackFixture { + public: + InProcess(Service* service, + const FixtureConfiguration& fixture_configuration = + FixtureConfiguration()) + : FullstackFixture(service, fixture_configuration, "") {} + ~InProcess() {} +}; + class EndpointPairFixture : public BaseFixture { public: EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints, @@ -279,6 +295,7 @@ class MinStackize : public Base { typedef MinStackize<TCP> MinTCP; typedef MinStackize<UDS> MinUDS; +typedef MinStackize<InProcess> MinInProcess; typedef MinStackize<SockPair> MinSockPair; typedef MinStackize<InProcessCHTTP2> MinInProcessCHTTP2; diff --git a/test/cpp/qps/BUILD b/test/cpp/qps/BUILD index 4d8b3b40e3..b3348b76fa 100644 --- a/test/cpp/qps/BUILD +++ b/test/cpp/qps/BUILD @@ -16,6 +16,13 @@ licenses(["notice"]) # Apache v2 load("//bazel:grpc_build_system.bzl", "grpc_cc_test", "grpc_cc_library", "grpc_cc_binary") +package( + features = [ + "-layering_check", + "-parse_headers", + ], +) + grpc_cc_library( name = "parse_json", srcs = ["parse_json.cc"], @@ -52,9 +59,6 @@ grpc_cc_library( "//test/core/util:grpc_test_util", "//test/cpp/util:test_util", ], - external_deps = [ - "gtest", - ], ) grpc_cc_library( @@ -77,6 +81,7 @@ grpc_cc_library( "//src/proto/grpc/testing:services_proto", "//test/core/util:gpr_test_util", "//test/core/util:grpc_test_util", + "//test/cpp/util:test_util", ], ) @@ -121,6 +126,7 @@ grpc_cc_binary( "//:gpr", "//test/core/util:gpr_test_util", "//test/core/util:grpc_test_util", + "//test/cpp/util:test_config", "//test/cpp/util:test_util", ], ) @@ -131,6 +137,7 @@ grpc_cc_test( deps = [ ":histogram", ":interarrival", + "//test/cpp/util:test_config", ], ) @@ -141,6 +148,8 @@ grpc_cc_binary( ":benchmark_config", ":driver_impl", "//:grpc++", + "//test/cpp/util:test_config", + "//test/cpp/util:test_util", ], external_deps = [ "gflags", @@ -154,6 +163,8 @@ grpc_cc_test( ":benchmark_config", ":driver_impl", ":qps_worker_impl", + "//test/cpp/util:test_config", + "//test/cpp/util:test_util", ], ) @@ -164,6 +175,8 @@ grpc_cc_test( ":benchmark_config", ":driver_impl", "//:grpc++", + "//test/cpp/util:test_config", + "//test/cpp/util:test_util", ], ) diff --git a/test/cpp/qps/benchmark_config.cc b/test/cpp/qps/benchmark_config.cc index 8f53670324..fb1e0608c5 100644 --- a/test/cpp/qps/benchmark_config.cc +++ b/test/cpp/qps/benchmark_config.cc @@ -54,10 +54,6 @@ using namespace gflags; namespace grpc { namespace testing { -void InitBenchmark(int* argc, char*** argv, bool remove_flags) { - ParseCommandLineFlags(argc, argv, remove_flags); -} - static std::shared_ptr<Reporter> InitBenchmarkReporters() { auto* composite_reporter = new CompositeReporter; if (FLAGS_enable_log_reporter) { diff --git a/test/cpp/qps/benchmark_config.h b/test/cpp/qps/benchmark_config.h index 054dbb276e..d3d6910ff0 100644 --- a/test/cpp/qps/benchmark_config.h +++ b/test/cpp/qps/benchmark_config.h @@ -20,15 +20,12 @@ #define GRPC_TEST_CPP_UTIL_BENCHMARK_CONFIG_H #include <memory> -#include <vector> #include "test/cpp/qps/report.h" namespace grpc { namespace testing { -void InitBenchmark(int* argc, char*** argv, bool remove_flags); - /** Returns the benchmark Reporter instance. * * The returned instance will take care of generating reports for all the actual diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index e759446c1a..6c4d92e859 100644 --- a/test/cpp/qps/client.h +++ b/test/cpp/qps/client.h @@ -39,6 +39,7 @@ #include "test/cpp/qps/interarrival.h" #include "test/cpp/qps/usage_timer.h" #include "test/cpp/util/create_test_channel.h" +#include "test/cpp/util/test_credentials_provider.h" namespace grpc { namespace testing { @@ -88,9 +89,7 @@ class ClientRequestCreator<ByteBuffer> { if (payload_config.has_bytebuf_params()) { std::unique_ptr<char[]> buf( new char[payload_config.bytebuf_params().req_size()]); - grpc_slice s = grpc_slice_from_copied_buffer( - buf.get(), payload_config.bytebuf_params().req_size()); - Slice slice(s, Slice::STEAL_REF); + Slice slice(buf.get(), payload_config.bytebuf_params().req_size()); *req = ByteBuffer(&slice, 1); } else { GPR_ASSERT(false); // not appropriate for this specialization @@ -407,9 +406,18 @@ class ClientImpl : public Client { ChannelArguments args; args.SetInt("shard_to_ensure_no_subchannel_merges", shard); set_channel_args(config, &args); + + grpc::string type; + if (config.has_security_params() && + config.security_params().cred_type().empty()) { + type = kTlsCredentialsType; + } else { + type = config.security_params().cred_type(); + } + channel_ = CreateTestChannel( - target, config.security_params().server_host_override(), - config.has_security_params(), !config.security_params().use_test_ca(), + target, type, config.security_params().server_host_override(), + !config.security_params().use_test_ca(), std::shared_ptr<CallCredentials>(), args); gpr_log(GPR_INFO, "Connecting to %s", target.c_str()); GPR_ASSERT(channel_->WaitForConnected( diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc index 6b934920df..5d212f1acc 100644 --- a/test/cpp/qps/client_sync.cc +++ b/test/cpp/qps/client_sync.cc @@ -140,7 +140,7 @@ class SynchronousStreamingClient : public SynchronousClient { if (*stream) { // forcibly cancel the streams, then finish context_[i].TryCancel(); - (*stream)->Finish(); + (*stream)->Finish().IgnoreError(); // don't log any error message on !ok since this was canceled } }); diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc index fbd8d1b1e7..4458e389e7 100644 --- a/test/cpp/qps/driver.cc +++ b/test/cpp/qps/driver.cc @@ -40,6 +40,7 @@ #include "test/cpp/qps/histogram.h" #include "test/cpp/qps/qps_worker.h" #include "test/cpp/qps/stats.h" +#include "test/cpp/util/test_credentials_provider.h" using std::list; using std::thread; @@ -172,13 +173,26 @@ static void postprocess_scenario_result(ScenarioResult* result) { sum(result->client_stats(), CliPollCount) / histogram.Count()); result->mutable_summary()->set_server_polls_per_request( sum(result->server_stats(), SvrPollCount) / histogram.Count()); + + auto server_queries_per_cpu_sec = + histogram.Count() / (sum(result->server_stats(), ServerSystemTime) + + sum(result->server_stats(), ServerUserTime)); + auto client_queries_per_cpu_sec = + histogram.Count() / (sum(result->client_stats(), SystemTime) + + sum(result->client_stats(), UserTime)); + + result->mutable_summary()->set_server_queries_per_cpu_sec( + server_queries_per_cpu_sec); + result->mutable_summary()->set_client_queries_per_cpu_sec( + client_queries_per_cpu_sec); } std::unique_ptr<ScenarioResult> RunScenario( const ClientConfig& initial_client_config, size_t num_clients, const ServerConfig& initial_server_config, size_t num_servers, int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count, - const char* qps_server_target_override) { + const grpc::string& qps_server_target_override, + const grpc::string& credential_type) { // Log everything from the driver gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG); @@ -214,7 +228,7 @@ std::unique_ptr<ScenarioResult> RunScenario( } int driver_port = grpc_pick_unused_port_or_die(); - local_workers.emplace_back(new QpsWorker(driver_port)); + local_workers.emplace_back(new QpsWorker(driver_port, 0, credential_type)); char addr[256]; sprintf(addr, "localhost:%d", driver_port); if (spawn_local_worker_count < 0) { @@ -246,12 +260,14 @@ std::unique_ptr<ScenarioResult> RunScenario( }; std::vector<ServerData> servers(num_servers); std::unordered_map<string, std::deque<int>> hosts_cores; + ChannelArguments channel_args; for (size_t i = 0; i < num_servers; i++) { gpr_log(GPR_INFO, "Starting server on %s (worker #%" PRIuPTR ")", workers[i].c_str(), i); - servers[i].stub = WorkerService::NewStub( - CreateChannel(workers[i], InsecureChannelCredentials())); + servers[i].stub = WorkerService::NewStub(CreateChannel( + workers[i], GetCredentialsProvider()->GetChannelCredentials( + credential_type, &channel_args))); ServerConfig server_config = initial_server_config; if (server_config.core_limit() != 0) { @@ -269,8 +285,7 @@ std::unique_ptr<ScenarioResult> RunScenario( if (!servers[i].stream->Read(&init_status)) { gpr_log(GPR_ERROR, "Server %zu did not yield initial status", i); } - if (qps_server_target_override != NULL && - strlen(qps_server_target_override) > 0) { + if (qps_server_target_override.length() > 0) { // overriding the qps server target only works if there is 1 server GPR_ASSERT(num_servers == 1); client_config.add_server_targets(qps_server_target_override); @@ -298,7 +313,8 @@ std::unique_ptr<ScenarioResult> RunScenario( gpr_log(GPR_INFO, "Starting client on %s (worker #%" PRIuPTR ")", worker.c_str(), i + num_servers); clients[i].stub = WorkerService::NewStub( - CreateChannel(worker, InsecureChannelCredentials())); + CreateChannel(worker, GetCredentialsProvider()->GetChannelCredentials( + credential_type, &channel_args))); ClientConfig per_client_config = client_config; if (initial_client_config.core_limit() != 0) { @@ -483,16 +499,19 @@ std::unique_ptr<ScenarioResult> RunScenario( return result; } -bool RunQuit() { +bool RunQuit(const grpc::string& credential_type) { // Get client, server lists bool result = true; auto workers = get_workers("QPS_WORKERS"); if (workers.size() == 0) { return false; } + + ChannelArguments channel_args; for (size_t i = 0; i < workers.size(); i++) { - auto stub = WorkerService::NewStub( - CreateChannel(workers[i], InsecureChannelCredentials())); + auto stub = WorkerService::NewStub(CreateChannel( + workers[i], GetCredentialsProvider()->GetChannelCredentials( + credential_type, &channel_args))); Void dummy; grpc::ClientContext ctx; ctx.set_wait_for_ready(true); diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h index def32c6f0e..29f2776d79 100644 --- a/test/cpp/qps/driver.h +++ b/test/cpp/qps/driver.h @@ -31,9 +31,10 @@ std::unique_ptr<ScenarioResult> RunScenario( const grpc::testing::ClientConfig& client_config, size_t num_clients, const grpc::testing::ServerConfig& server_config, size_t num_servers, int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count, - const char* qps_server_target_override = ""); + const grpc::string& qps_server_target_override, + const grpc::string& credential_type); -bool RunQuit(); +bool RunQuit(const grpc::string& credential_type); } // namespace testing } // namespace grpc diff --git a/test/cpp/qps/qps_interarrival_test.cc b/test/cpp/qps/qps_interarrival_test.cc index d19266ff9c..87f09e8c5f 100644 --- a/test/cpp/qps/qps_interarrival_test.cc +++ b/test/cpp/qps/qps_interarrival_test.cc @@ -23,6 +23,7 @@ #include <grpc/support/histogram.h> #include "test/cpp/qps/interarrival.h" +#include "test/cpp/util/test_config.h" using grpc::testing::RandomDistInterface; using grpc::testing::InterarrivalTimer; @@ -50,6 +51,8 @@ static void RunTest(RandomDistInterface &&r, int threads, std::string title) { using grpc::testing::ExpDist; int main(int argc, char **argv) { + grpc::testing::InitTest(&argc, &argv, true); + RunTest(ExpDist(10.0), 5, std::string("Exponential(10)")); return 0; } diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc index 590c22ec29..cca59f64d8 100644 --- a/test/cpp/qps/qps_json_driver.cc +++ b/test/cpp/qps/qps_json_driver.cc @@ -30,6 +30,8 @@ #include "test/cpp/qps/driver.h" #include "test/cpp/qps/parse_json.h" #include "test/cpp/qps/report.h" +#include "test/cpp/util/test_config.h" +#include "test/cpp/util/test_credentials_provider.h" DEFINE_string(scenarios_file, "", "JSON file containing an array of Scenario objects"); @@ -60,6 +62,9 @@ DEFINE_string(qps_server_target_override, "", DEFINE_string(json_file_out, "", "File to write the JSON output to."); +DEFINE_string(credential_type, grpc::testing::kInsecureCredentialsType, + "Credential type for communication with workers"); + namespace grpc { namespace testing { @@ -71,7 +76,7 @@ static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario, scenario.server_config(), scenario.num_servers(), scenario.warmup_seconds(), scenario.benchmark_seconds(), scenario.spawn_local_worker_count(), - FLAGS_qps_server_target_override.c_str()); + FLAGS_qps_server_target_override, FLAGS_credential_type); // Amend the result with scenario config. Eventually we should adjust // RunScenario contract so we don't need to touch the result here. @@ -83,6 +88,7 @@ static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario, GetReporter()->ReportTimes(*result); GetReporter()->ReportCpuUsage(*result); GetReporter()->ReportPollCount(*result); + GetReporter()->ReportQueriesPerCpuSec(*result); for (int i = 0; *success && i < result->client_success_size(); i++) { *success = result->client_success(i); @@ -184,7 +190,7 @@ static bool QpsDriver() { } else if (scjson) { json = FLAGS_scenarios_json.c_str(); } else if (FLAGS_quit) { - return RunQuit(); + return RunQuit(FLAGS_credential_type); } // Parse into an array of scenarios @@ -219,7 +225,7 @@ static bool QpsDriver() { } // namespace grpc int main(int argc, char** argv) { - grpc::testing::InitBenchmark(&argc, &argv, true); + grpc::testing::InitTest(&argc, &argv, true); bool ok = grpc::testing::QpsDriver(); diff --git a/test/cpp/qps/qps_openloop_test.cc b/test/cpp/qps/qps_openloop_test.cc index 28dec4f8c6..069b3fa076 100644 --- a/test/cpp/qps/qps_openloop_test.cc +++ b/test/cpp/qps/qps_openloop_test.cc @@ -24,6 +24,8 @@ #include "test/cpp/qps/benchmark_config.h" #include "test/cpp/qps/driver.h" #include "test/cpp/qps/report.h" +#include "test/cpp/util/test_config.h" +#include "test/cpp/util/test_credentials_provider.h" namespace grpc { namespace testing { @@ -47,8 +49,8 @@ static void RunQPS() { server_config.set_server_type(ASYNC_SERVER); server_config.set_async_server_threads(8); - const auto result = - RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); + const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, + BENCHMARK, -2, "", kInsecureCredentialsType); GetReporter()->ReportQPSPerCore(*result); GetReporter()->ReportLatency(*result); @@ -58,7 +60,7 @@ static void RunQPS() { } // namespace grpc int main(int argc, char** argv) { - grpc::testing::InitBenchmark(&argc, &argv, true); + grpc::testing::InitTest(&argc, &argv, true); grpc::testing::RunQPS(); diff --git a/test/cpp/qps/qps_test.cc b/test/cpp/qps/qps_test.cc deleted file mode 100644 index 6ab2102e40..0000000000 --- a/test/cpp/qps/qps_test.cc +++ /dev/null @@ -1,64 +0,0 @@ -/* - * - * Copyright 2015 gRPC authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -#include <set> - -#include <grpc/support/log.h> - -#include "test/cpp/qps/benchmark_config.h" -#include "test/cpp/qps/driver.h" -#include "test/cpp/qps/report.h" - -namespace grpc { -namespace testing { - -static const int WARMUP = 20; -static const int BENCHMARK = 20; - -static void RunQPS() { - gpr_log(GPR_INFO, "Running QPS test"); - - ClientConfig client_config; - client_config.set_client_type(ASYNC_CLIENT); - client_config.set_outstanding_rpcs_per_channel(100); - client_config.set_client_channels(64); - client_config.set_async_client_threads(8); - client_config.set_rpc_type(STREAMING); - client_config.mutable_load_params()->mutable_closed_loop(); - - ServerConfig server_config; - server_config.set_server_type(ASYNC_SERVER); - server_config.set_async_server_threads(8); - - const auto result = - RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); - - GetReporter()->ReportQPSPerCore(*result); - GetReporter()->ReportLatency(*result); -} - -} // namespace testing -} // namespace grpc - -int main(int argc, char** argv) { - grpc::testing::InitBenchmark(&argc, &argv, true); - - grpc::testing::RunQPS(); - - return 0; -} diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc index 10bc5422e1..d20bc1b074 100644 --- a/test/cpp/qps/qps_worker.cc +++ b/test/cpp/qps/qps_worker.cc @@ -41,6 +41,7 @@ #include "test/cpp/qps/client.h" #include "test/cpp/qps/server.h" #include "test/cpp/util/create_test_channel.h" +#include "test/cpp/util/test_credentials_provider.h" namespace grpc { namespace testing { @@ -263,7 +264,8 @@ class WorkerServiceImpl final : public WorkerService::Service { QpsWorker* worker_; }; -QpsWorker::QpsWorker(int driver_port, int server_port) { +QpsWorker::QpsWorker(int driver_port, int server_port, + const grpc::string& credential_type) { impl_.reset(new WorkerServiceImpl(server_port, this)); gpr_atm_rel_store(&done_, static_cast<gpr_atm>(0)); @@ -271,7 +273,9 @@ QpsWorker::QpsWorker(int driver_port, int server_port) { gpr_join_host_port(&server_address, "::", driver_port); ServerBuilder builder; - builder.AddListeningPort(server_address, InsecureServerCredentials()); + builder.AddListeningPort( + server_address, + GetCredentialsProvider()->GetServerCredentials(credential_type)); builder.RegisterService(impl_.get()); gpr_free(server_address); diff --git a/test/cpp/qps/qps_worker.h b/test/cpp/qps/qps_worker.h index c8a7be9360..360125fb17 100644 --- a/test/cpp/qps/qps_worker.h +++ b/test/cpp/qps/qps_worker.h @@ -21,6 +21,7 @@ #include <memory> +#include <grpc++/support/config.h> #include <grpc/support/atm.h> namespace grpc { @@ -33,7 +34,8 @@ class WorkerServiceImpl; class QpsWorker { public: - explicit QpsWorker(int driver_port, int server_port = 0); + explicit QpsWorker(int driver_port, int server_port, + const grpc::string& credential_type); ~QpsWorker(); bool Done() const; diff --git a/test/cpp/qps/report.cc b/test/cpp/qps/report.cc index 809c563044..a45b10bcb8 100644 --- a/test/cpp/qps/report.cc +++ b/test/cpp/qps/report.cc @@ -71,6 +71,12 @@ void CompositeReporter::ReportPollCount(const ScenarioResult& result) { } } +void CompositeReporter::ReportQueriesPerCpuSec(const ScenarioResult& result) { + for (size_t i = 0; i < reporters_.size(); ++i) { + reporters_[i]->ReportQueriesPerCpuSec(result); + } +} + void GprLogReporter::ReportQPS(const ScenarioResult& result) { gpr_log(GPR_INFO, "QPS: %.1f", result.summary().qps()); if (result.summary().failed_requests_per_second() > 0) { @@ -119,6 +125,13 @@ void GprLogReporter::ReportPollCount(const ScenarioResult& result) { result.summary().server_polls_per_request()); } +void GprLogReporter::ReportQueriesPerCpuSec(const ScenarioResult& result) { + gpr_log(GPR_INFO, "Server Queries/CPU-sec: %.2f", + result.summary().server_queries_per_cpu_sec()); + gpr_log(GPR_INFO, "Client Queries/CPU-sec: %.2f", + result.summary().client_queries_per_cpu_sec()); +} + void JsonReporter::ReportQPS(const ScenarioResult& result) { grpc::string json_string = SerializeJson(result, "type.googleapis.com/grpc.testing.ScenarioResult"); @@ -147,6 +160,10 @@ void JsonReporter::ReportPollCount(const ScenarioResult& result) { // NOP - all reporting is handled by ReportQPS. } +void JsonReporter::ReportQueriesPerCpuSec(const ScenarioResult& result) { + // NOP - all reporting is handled by ReportQPS. +} + void RpcReporter::ReportQPS(const ScenarioResult& result) { grpc::ClientContext context; grpc::Status status; @@ -183,5 +200,9 @@ void RpcReporter::ReportPollCount(const ScenarioResult& result) { // NOP - all reporting is handled by ReportQPS. } +void RpcReporter::ReportQueriesPerCpuSec(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 0bd398fd2a..321be2a97f 100644 --- a/test/cpp/qps/report.h +++ b/test/cpp/qps/report.h @@ -64,6 +64,9 @@ class Reporter { /** Reports client and server poll usage inside completion queue. */ virtual void ReportPollCount(const ScenarioResult& result) = 0; + /** Reports queries per cpu-sec. */ + virtual void ReportQueriesPerCpuSec(const ScenarioResult& result) = 0; + private: const string name_; }; @@ -82,6 +85,7 @@ class CompositeReporter : public Reporter { void ReportTimes(const ScenarioResult& result) override; void ReportCpuUsage(const ScenarioResult& result) override; void ReportPollCount(const ScenarioResult& result) override; + void ReportQueriesPerCpuSec(const ScenarioResult& result) override; private: std::vector<std::unique_ptr<Reporter> > reporters_; @@ -99,6 +103,7 @@ class GprLogReporter : public Reporter { void ReportTimes(const ScenarioResult& result) override; void ReportCpuUsage(const ScenarioResult& result) override; void ReportPollCount(const ScenarioResult& result) override; + void ReportQueriesPerCpuSec(const ScenarioResult& result) override; }; /** Dumps the report to a JSON file. */ @@ -114,6 +119,7 @@ class JsonReporter : public Reporter { void ReportTimes(const ScenarioResult& result) override; void ReportCpuUsage(const ScenarioResult& result) override; void ReportPollCount(const ScenarioResult& result) override; + void ReportQueriesPerCpuSec(const ScenarioResult& result) override; const string report_file_; }; @@ -130,6 +136,7 @@ class RpcReporter : public Reporter { void ReportTimes(const ScenarioResult& result) override; void ReportCpuUsage(const ScenarioResult& result) override; void ReportPollCount(const ScenarioResult& result) override; + void ReportQueriesPerCpuSec(const ScenarioResult& result) override; std::unique_ptr<ReportQpsScenarioService::Stub> stub_; }; diff --git a/test/cpp/qps/secure_sync_unary_ping_pong_test.cc b/test/cpp/qps/secure_sync_unary_ping_pong_test.cc index 586392692e..137b33ee25 100644 --- a/test/cpp/qps/secure_sync_unary_ping_pong_test.cc +++ b/test/cpp/qps/secure_sync_unary_ping_pong_test.cc @@ -23,6 +23,8 @@ #include "test/cpp/qps/benchmark_config.h" #include "test/cpp/qps/driver.h" #include "test/cpp/qps/report.h" +#include "test/cpp/util/test_config.h" +#include "test/cpp/util/test_credentials_provider.h" namespace grpc { namespace testing { @@ -50,8 +52,8 @@ static void RunSynchronousUnaryPingPong() { client_config.mutable_security_params()->CopyFrom(security); server_config.mutable_security_params()->CopyFrom(security); - const auto result = - RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); + const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, + BENCHMARK, -2, "", kInsecureCredentialsType); GetReporter()->ReportQPS(*result); GetReporter()->ReportLatency(*result); @@ -61,7 +63,7 @@ static void RunSynchronousUnaryPingPong() { } // namespace grpc int main(int argc, char** argv) { - grpc::testing::InitBenchmark(&argc, &argv, true); + grpc::testing::InitTest(&argc, &argv, true); grpc::testing::RunSynchronousUnaryPingPong(); diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h index a3b4d7d241..c0dac96d8b 100644 --- a/test/cpp/qps/server.h +++ b/test/cpp/qps/server.h @@ -19,8 +19,11 @@ #ifndef TEST_QPS_SERVER_H #define TEST_QPS_SERVER_H +#include <grpc++/resource_quota.h> #include <grpc++/security/server_credentials.h> +#include <grpc++/server_builder.h> #include <grpc/support/cpu.h> +#include <grpc/support/log.h> #include <vector> #include "src/core/lib/surface/completion_queue.h" @@ -29,6 +32,7 @@ #include "test/core/end2end/data/ssl_test_data.h" #include "test/core/util/port.h" #include "test/cpp/qps/usage_timer.h" +#include "test/cpp/util/test_credentials_provider.h" namespace grpc { namespace testing { @@ -86,12 +90,14 @@ class Server { static std::shared_ptr<ServerCredentials> CreateServerCredentials( const ServerConfig& config) { if (config.has_security_params()) { - SslServerCredentialsOptions::PemKeyCertPair pkcp = {test_server1_key, - test_server1_cert}; - SslServerCredentialsOptions ssl_opts; - ssl_opts.pem_root_certs = ""; - ssl_opts.pem_key_cert_pairs.push_back(pkcp); - return SslServerCredentials(ssl_opts); + grpc::string type; + if (config.security_params().cred_type().empty()) { + type = kTlsCredentialsType; + } else { + type = config.security_params().cred_type(); + } + + return GetCredentialsProvider()->GetServerCredentials(type); } else { return InsecureServerCredentials(); } @@ -102,6 +108,31 @@ class Server { return 0; } + protected: + static void ApplyConfigToBuilder(const ServerConfig& config, + ServerBuilder* builder) { + if (config.resource_quota_size() > 0) { + builder->SetResourceQuota(ResourceQuota("AsyncQpsServerTest") + .Resize(config.resource_quota_size())); + } + for (const auto& channel_arg : config.channel_args()) { + switch (channel_arg.value_case()) { + case ChannelArg::kStrValue: + builder->AddChannelArgument(channel_arg.name(), + channel_arg.str_value()); + break; + case ChannelArg::kIntValue: + builder->AddChannelArgument(channel_arg.name(), + channel_arg.int_value()); + break; + case ChannelArg::VALUE_NOT_SET: + gpr_log(GPR_ERROR, "Channel arg '%s' does not have a value", + channel_arg.name().c_str()); + break; + } + } + } + private: int port_; int cores_; diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc index 96d7e5ef74..122976c397 100644 --- a/test/cpp/qps/server_async.cc +++ b/test/cpp/qps/server_async.cc @@ -99,10 +99,7 @@ class AsyncQpsServerTest final : public grpc::testing::Server { cq_.emplace_back(i % srv_cqs_.size()); } - if (config.resource_quota_size() > 0) { - builder.SetResourceQuota(ResourceQuota("AsyncQpsServerTest") - .Resize(config.resource_quota_size())); - } + ApplyConfigToBuilder(config, &builder); server_ = builder.BuildAndStart(); diff --git a/test/cpp/qps/server_sync.cc b/test/cpp/qps/server_sync.cc index 90aa89491f..9954e2c0bf 100644 --- a/test/cpp/qps/server_sync.cc +++ b/test/cpp/qps/server_sync.cc @@ -19,15 +19,12 @@ #include <atomic> #include <thread> -#include <grpc++/resource_quota.h> #include <grpc++/security/server_credentials.h> #include <grpc++/server.h> -#include <grpc++/server_builder.h> #include <grpc++/server_context.h> #include <grpc/grpc.h> #include <grpc/support/alloc.h> #include <grpc/support/host_port.h> -#include <grpc/support/log.h> #include "src/proto/grpc/testing/services.grpc.pb.h" #include "test/cpp/qps/server.h" @@ -115,8 +112,8 @@ class BenchmarkServiceImpl final : public BenchmarkService::Service { } private: - static Status ClientPull(ServerContext* context, - ReaderInterface<SimpleRequest>* stream, + template <class R> + static Status ClientPull(ServerContext* context, R* stream, SimpleResponse* response) { SimpleRequest request; while (stream->Read(&request)) { @@ -129,8 +126,8 @@ class BenchmarkServiceImpl final : public BenchmarkService::Service { } return Status::OK; } - static Status ServerPush(ServerContext* context, - WriterInterface<SimpleResponse>* stream, + template <class W> + static Status ServerPush(ServerContext* context, W* stream, const SimpleResponse& response, std::function<bool()> done) { while ((done == nullptr) || !done()) { @@ -166,10 +163,7 @@ class SynchronousServer final : public grpc::testing::Server { Server::CreateServerCredentials(config)); gpr_free(server_address); - if (config.resource_quota_size() > 0) { - builder.SetResourceQuota(ResourceQuota("AsyncQpsServerTest") - .Resize(config.resource_quota_size())); - } + ApplyConfigToBuilder(config, &builder); builder.RegisterService(&service_); diff --git a/test/cpp/qps/worker.cc b/test/cpp/qps/worker.cc index fd51d32db0..27010b7315 100644 --- a/test/cpp/qps/worker.cc +++ b/test/cpp/qps/worker.cc @@ -27,9 +27,12 @@ #include "test/cpp/qps/qps_worker.h" #include "test/cpp/util/test_config.h" +#include "test/cpp/util/test_credentials_provider.h" DEFINE_int32(driver_port, 0, "Port for communication with driver"); DEFINE_int32(server_port, 0, "Port for operation as a server"); +DEFINE_string(credential_type, grpc::testing::kInsecureCredentialsType, + "Credential type for communication with driver"); static bool got_sigint = false; @@ -39,7 +42,7 @@ namespace grpc { namespace testing { static void RunServer() { - QpsWorker worker(FLAGS_driver_port, FLAGS_server_port); + QpsWorker worker(FLAGS_driver_port, FLAGS_server_port, FLAGS_credential_type); while (!got_sigint && !worker.Done()) { gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), diff --git a/test/cpp/server/BUILD b/test/cpp/server/BUILD new file mode 100644 index 0000000000..512241e350 --- /dev/null +++ b/test/cpp/server/BUILD @@ -0,0 +1,43 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +licenses(["notice"]) # Apache v2 + +load("//bazel:grpc_build_system.bzl", "grpc_cc_test", "grpc_cc_library", "grpc_cc_binary") + +grpc_cc_test( + name = "server_builder_test", + srcs = ["server_builder_test.cc"], + deps = [ + "//:grpc++", + "//src/proto/grpc/testing:echo_proto", + "//test/core/util:grpc_test_util", + ], + external_deps = [ + "gtest", + ], +) + +grpc_cc_test( + name = "server_request_call_test", + srcs = ["server_request_call_test.cc"], + deps = [ + "//:grpc++", + "//src/proto/grpc/testing:echo_proto", + "//test/core/util:grpc_test_util", + ], + external_deps = [ + "gtest", + ], +) diff --git a/test/cpp/server/server_request_call_test.cc b/test/cpp/server/server_request_call_test.cc new file mode 100644 index 0000000000..b0a2fa444c --- /dev/null +++ b/test/cpp/server/server_request_call_test.cc @@ -0,0 +1,163 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include <thread> + +#include <grpc++/impl/codegen/config.h> + +#include <grpc++/server.h> +#include <grpc++/server_builder.h> + +#include <grpc++/create_channel.h> +#include <grpc++/security/credentials.h> + +#include <grpc/support/log.h> + +#include "src/proto/grpc/testing/echo.grpc.pb.h" +#include "test/core/util/port.h" + +#include <gtest/gtest.h> + +namespace grpc { +namespace { + +TEST(ServerRequestCallTest, ShortDeadlineDoesNotCauseOkayFalse) { + std::mutex mu; + bool shutting_down = false; + + // grpc server config. + std::ostringstream s; + int p = grpc_pick_unused_port_or_die(); + s << "[::1]:" << p; + const string address = s.str(); + testing::EchoTestService::AsyncService service; + ServerBuilder builder; + builder.AddListeningPort(address, InsecureServerCredentials()); + auto cq = builder.AddCompletionQueue(); + builder.RegisterService(&service); + auto server = builder.BuildAndStart(); + + // server thread. + std::thread t([address, &service, &cq, &mu, &shutting_down] { + for (int n = 0; true; n++) { + ServerContext ctx; + testing::EchoRequest req; + ServerAsyncResponseWriter<testing::EchoResponse> responder(&ctx); + + // if shutting down, don't enqueue a new request. + { + std::lock_guard<std::mutex> lock(mu); + if (!shutting_down) { + service.RequestEcho(&ctx, &req, &responder, cq.get(), cq.get(), + (void*)1); + } + } + + bool ok; + void* tag; + if (!cq->Next(&tag, &ok)) { + break; + } + + EXPECT_EQ((void*)1, tag); + // If not shutting down, ok must be true for new requests. + { + std::lock_guard<std::mutex> lock(mu); + if (!shutting_down && !ok) { + gpr_log(GPR_INFO, "!ok on request %d", n); + abort(); + } + if (shutting_down && !ok) { + // Failed connection due to shutdown, continue flushing the CQ. + continue; + } + } + + // Send a simple response after a small delay that would ensure the client + // deadline is exceeded. + gpr_log(GPR_INFO, "Got request %d", n); + testing::EchoResponse response; + response.set_message("foobar"); + // A bit of sleep to make sure the deadline elapses. + gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), + gpr_time_from_millis(50, GPR_TIMESPAN))); + { + std::lock_guard<std::mutex> lock(mu); + if (shutting_down) { + gpr_log(GPR_INFO, + "shut down while processing call, not calling Finish()"); + // Continue flushing the CQ. + continue; + } + gpr_log(GPR_INFO, "Finishing request %d", n); + responder.Finish(response, grpc::Status::OK, (void*)2); + if (!cq->Next(&tag, &ok)) { + break; + } + EXPECT_EQ((void*)2, tag); + } + } + }); + + auto stub = testing::EchoTestService::NewStub( + CreateChannel(address, InsecureChannelCredentials())); + + for (int i = 0; i < 100; i++) { + gpr_log(GPR_INFO, "Sending %d.", i); + testing::EchoRequest request; + + ///////// + // Comment out the following line to get ok=false due to invalid request. + // Otherwise, ok=false due to deadline being exceeded. + ///////// + request.set_message("foobar"); + + // A simple request with a short deadline. The server will always exceed the + // deadline, whether due to the sleep or because the server was unable to + // even fetch the request from the CQ before the deadline elapsed. + testing::EchoResponse response; + ::grpc::ClientContext ctx; + ctx.set_fail_fast(false); + ctx.set_deadline(std::chrono::system_clock::now() + + std::chrono::milliseconds(1)); + grpc::Status status = stub->Echo(&ctx, request, &response); + EXPECT_EQ(DEADLINE_EXCEEDED, status.error_code()); + gpr_log(GPR_INFO, "Success."); + } + gpr_log(GPR_INFO, "Done sending RPCs."); + + // Shut down everything properly. + gpr_log(GPR_INFO, "Shutting down."); + { + std::lock_guard<std::mutex> lock(mu); + shutting_down = true; + } + server->Shutdown(); + cq->Shutdown(); + server->Wait(); + + t.join(); +} + +} // namespace +} // namespace grpc + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/cpp/util/BUILD b/test/cpp/util/BUILD index 6db61abfda..33240f6f69 100644 --- a/test/cpp/util/BUILD +++ b/test/cpp/util/BUILD @@ -14,9 +14,15 @@ licenses(["notice"]) # Apache v2 -load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_binary") +load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_binary", "grpc_cc_test") -package(default_visibility = ["//visibility:public"]) +package( + default_visibility = ["//visibility:public"], + features = [ + "-layering_check", + "-parse_headers", + ], +) # The following builds a shared-object to confirm that grpc++_unsecure # builds properly. Build-only is sufficient here @@ -126,19 +132,21 @@ grpc_cc_library( ], ) -cc_test( +grpc_cc_test( name = "error_details_test", srcs = [ "error_details_test.cc", ], deps = [ "//:grpc++_error_details", - "//external:gtest", "//src/proto/grpc/testing:echo_messages_proto", ], + external_deps = [ + "gtest", + ], ) -cc_binary( +grpc_cc_binary( name = "grpc_cli", srcs = [ "cli_call.cc", @@ -160,7 +168,9 @@ cc_binary( ], deps = [ "//:grpc++", - "//external:gflags", "//src/proto/grpc/reflection/v1alpha:reflection_proto", ], + external_deps = [ + "gflags", + ], ) diff --git a/test/cpp/util/byte_buffer_proto_helper.cc b/test/cpp/util/byte_buffer_proto_helper.cc index 22dd074fe9..bb5162f86e 100644 --- a/test/cpp/util/byte_buffer_proto_helper.cc +++ b/test/cpp/util/byte_buffer_proto_helper.cc @@ -36,8 +36,7 @@ std::unique_ptr<ByteBuffer> SerializeToByteBuffer( grpc::protobuf::Message* message) { grpc::string buf; message->SerializeToString(&buf); - grpc_slice s = grpc_slice_from_copied_string(buf.c_str()); - Slice slice(s, Slice::STEAL_REF); + Slice slice(buf); return std::unique_ptr<ByteBuffer>(new ByteBuffer(&slice, 1)); } diff --git a/test/cpp/util/byte_buffer_test.cc b/test/cpp/util/byte_buffer_test.cc index 6c077ddb2c..cac01a7307 100644 --- a/test/cpp/util/byte_buffer_test.cc +++ b/test/cpp/util/byte_buffer_test.cc @@ -34,33 +34,30 @@ const char* kContent2 = "yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy world"; class ByteBufferTest : public ::testing::Test {}; TEST_F(ByteBufferTest, CreateFromSingleSlice) { - grpc_slice hello = grpc_slice_from_copied_string(kContent1); - Slice s(hello, Slice::STEAL_REF); + Slice s(kContent1); ByteBuffer buffer(&s, 1); + EXPECT_EQ(strlen(kContent1), buffer.Length()); } TEST_F(ByteBufferTest, CreateFromVector) { - grpc_slice hello = grpc_slice_from_copied_string(kContent1); - grpc_slice world = grpc_slice_from_copied_string(kContent2); std::vector<Slice> slices; - slices.push_back(Slice(hello, Slice::STEAL_REF)); - slices.push_back(Slice(world, Slice::STEAL_REF)); + slices.emplace_back(kContent1); + slices.emplace_back(kContent2); ByteBuffer buffer(&slices[0], 2); + EXPECT_EQ(strlen(kContent1) + strlen(kContent2), buffer.Length()); } TEST_F(ByteBufferTest, Clear) { - grpc_slice hello = grpc_slice_from_copied_string(kContent1); - Slice s(hello, Slice::STEAL_REF); + Slice s(kContent1); ByteBuffer buffer(&s, 1); buffer.Clear(); + EXPECT_EQ(static_cast<size_t>(0), buffer.Length()); } TEST_F(ByteBufferTest, Length) { - grpc_slice hello = grpc_slice_from_copied_string(kContent1); - grpc_slice world = grpc_slice_from_copied_string(kContent2); std::vector<Slice> slices; - slices.push_back(Slice(hello, Slice::STEAL_REF)); - slices.push_back(Slice(world, Slice::STEAL_REF)); + slices.emplace_back(kContent1); + slices.emplace_back(kContent2); ByteBuffer buffer(&slices[0], 2); EXPECT_EQ(strlen(kContent1) + strlen(kContent2), buffer.Length()); } diff --git a/test/cpp/util/cli_call.cc b/test/cpp/util/cli_call.cc index fb6d76bb90..c3220efa54 100644 --- a/test/cpp/util/cli_call.cc +++ b/test/cpp/util/cli_call.cc @@ -119,8 +119,7 @@ void CliCall::WritesDone() { } void CliCall::WriteAndWait(const grpc::string& request) { - grpc_slice s = grpc_slice_from_copied_string(request.c_str()); - grpc::Slice req_slice(s, grpc::Slice::STEAL_REF); + grpc::Slice req_slice(request); grpc::ByteBuffer send_buffer(&req_slice, 1); gpr_mu_lock(&write_mu_); diff --git a/test/cpp/util/create_test_channel.cc b/test/cpp/util/create_test_channel.cc index 68c6fe4622..34b6d60d01 100644 --- a/test/cpp/util/create_test_channel.cc +++ b/test/cpp/util/create_test_channel.cc @@ -51,29 +51,31 @@ void AddProdSslType() { } // namespace -// When ssl is enabled, if server is empty, override_hostname is used to +// When cred_type is 'ssl', if server is empty, override_hostname is used to // create channel. Otherwise, connect to server and override hostname if // override_hostname is provided. -// When ssl is not enabled, override_hostname is ignored. +// When cred_type is not 'ssl', override_hostname is ignored. // Set use_prod_root to true to use the SSL root for connecting to google. // In this case, path to the roots pem file must be set via environment variable // GRPC_DEFAULT_SSL_ROOTS_FILE_PATH. // Otherwise, root for test SSL cert will be used. -// creds will be used to create a channel when enable_ssl is true. +// creds will be used to create a channel when cred_type is 'ssl'. // Use examples: // CreateTestChannel( -// "1.1.1.1:12345", "override.hostname.com", true, false, creds); -// CreateTestChannel("test.google.com:443", "", true, true, creds); +// "1.1.1.1:12345", "ssl", "override.hostname.com", false, creds); +// CreateTestChannel("test.google.com:443", "ssl", "", true, creds); // same as above -// CreateTestChannel("", "test.google.com:443", true, true, creds); +// CreateTestChannel("", "ssl", "test.google.com:443", true, creds); std::shared_ptr<Channel> CreateTestChannel( - const grpc::string& server, const grpc::string& override_hostname, - bool enable_ssl, bool use_prod_roots, + const grpc::string& server, const grpc::string& cred_type, + const grpc::string& override_hostname, bool use_prod_roots, const std::shared_ptr<CallCredentials>& creds, const ChannelArguments& args) { ChannelArguments channel_args(args); std::shared_ptr<ChannelCredentials> channel_creds; - if (enable_ssl) { + if (cred_type.empty()) { + return CreateChannel(server, InsecureChannelCredentials()); + } else if (cred_type == testing::kTlsCredentialsType) { // cred_type == "ssl" if (use_prod_roots) { gpr_once_init(&g_once_init_add_prod_ssl_provider, &AddProdSslType); channel_creds = testing::GetCredentialsProvider()->GetChannelCredentials( @@ -95,13 +97,31 @@ std::shared_ptr<Channel> CreateTestChannel( } return CreateCustomChannel(connect_to, channel_creds, channel_args); } else { - return CreateChannel(server, InsecureChannelCredentials()); + channel_creds = testing::GetCredentialsProvider()->GetChannelCredentials( + cred_type, &channel_args); + GPR_ASSERT(channel_creds != nullptr); + + return CreateChannel(server, channel_creds); } } std::shared_ptr<Channel> CreateTestChannel( const grpc::string& server, const grpc::string& override_hostname, bool enable_ssl, bool use_prod_roots, + const std::shared_ptr<CallCredentials>& creds, + const ChannelArguments& args) { + grpc::string type; + if (enable_ssl) { + type = testing::kTlsCredentialsType; + } + + return CreateTestChannel(server, type, override_hostname, use_prod_roots, + creds, args); +} + +std::shared_ptr<Channel> CreateTestChannel( + const grpc::string& server, const grpc::string& override_hostname, + bool enable_ssl, bool use_prod_roots, const std::shared_ptr<CallCredentials>& creds) { return CreateTestChannel(server, override_hostname, enable_ssl, use_prod_roots, creds, ChannelArguments()); diff --git a/test/cpp/util/create_test_channel.h b/test/cpp/util/create_test_channel.h index 9b4b09171e..e2ca8f99b4 100644 --- a/test/cpp/util/create_test_channel.h +++ b/test/cpp/util/create_test_channel.h @@ -45,6 +45,12 @@ std::shared_ptr<Channel> CreateTestChannel( const ChannelArguments& args); std::shared_ptr<Channel> CreateTestChannel( + const grpc::string& server, const grpc::string& cred_type, + const grpc::string& override_hostname, bool use_prod_roots, + const std::shared_ptr<CallCredentials>& creds, + const ChannelArguments& args); + +std::shared_ptr<Channel> CreateTestChannel( const grpc::string& server, const grpc::string& credential_type, const std::shared_ptr<CallCredentials>& creds); diff --git a/test/cpp/util/grpc_tool_test.cc b/test/cpp/util/grpc_tool_test.cc index 4069c7a91a..dd00581f2b 100644 --- a/test/cpp/util/grpc_tool_test.cc +++ b/test/cpp/util/grpc_tool_test.cc @@ -87,7 +87,7 @@ DECLARE_bool(l); namespace { -const int kNumResponseStreamsMsgs = 3; +const int kServerDefaultResponseStreamsToSend = 3; class TestCliCredentials final : public grpc::testing::CliCredentials { public: @@ -159,7 +159,7 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service { context->AddTrailingMetadata("trailing_key", "trailing_value"); EchoResponse response; - for (int i = 0; i < kNumResponseStreamsMsgs; i++) { + for (int i = 0; i < kServerDefaultResponseStreamsToSend; i++) { response.set_message(request->message() + grpc::to_string(i)); writer->Write(response); } @@ -463,7 +463,7 @@ TEST_F(GrpcToolTest, CallCommandResponseStream) { std::placeholders::_1))); // Expected output: "message: \"Hello{n}\"" - for (int i = 0; i < kNumResponseStreamsMsgs; i++) { + for (int i = 0; i < kServerDefaultResponseStreamsToSend; i++) { grpc::string expected_response_text = "message: \"Hello" + grpc::to_string(i) + "\"\n"; EXPECT_TRUE(NULL != strstr(output_stream.str().c_str(), diff --git a/test/cpp/util/slice_test.cc b/test/cpp/util/slice_test.cc index 0f80003580..9e3329fab0 100644 --- a/test/cpp/util/slice_test.cc +++ b/test/cpp/util/slice_test.cc @@ -28,6 +28,9 @@ const char* kContent = "hello xxxxxxxxxxxxxxxxxxxx world"; class SliceTest : public ::testing::Test { protected: + void CheckSliceSize(const Slice& s, const grpc::string& content) { + EXPECT_EQ(content.size(), s.size()); + } void CheckSlice(const Slice& s, const grpc::string& content) { EXPECT_EQ(content.size(), s.size()); EXPECT_EQ(content, @@ -35,6 +38,31 @@ class SliceTest : public ::testing::Test { } }; +TEST_F(SliceTest, Empty) { + Slice empty_slice; + CheckSlice(empty_slice, ""); +} + +TEST_F(SliceTest, Sized) { + Slice sized_slice(strlen(kContent)); + CheckSliceSize(sized_slice, kContent); +} + +TEST_F(SliceTest, String) { + Slice spp(kContent); + CheckSlice(spp, kContent); +} + +TEST_F(SliceTest, Buf) { + Slice spp(kContent, strlen(kContent)); + CheckSlice(spp, kContent); +} + +TEST_F(SliceTest, StaticBuf) { + Slice spp(kContent, strlen(kContent), Slice::STATIC_SLICE); + CheckSlice(spp, kContent); +} + TEST_F(SliceTest, Steal) { grpc_slice s = grpc_slice_from_copied_string(kContent); Slice spp(s, Slice::STEAL_REF); @@ -48,11 +76,6 @@ TEST_F(SliceTest, Add) { CheckSlice(spp, kContent); } -TEST_F(SliceTest, Empty) { - Slice empty_slice; - CheckSlice(empty_slice, ""); -} - TEST_F(SliceTest, Cslice) { grpc_slice s = grpc_slice_from_copied_string(kContent); Slice spp(s, Slice::STEAL_REF); |