aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp')
-rw-r--r--test/cpp/codegen/compiler_test_golden294
-rw-r--r--test/cpp/codegen/golden_file_test.cc64
-rw-r--r--test/cpp/end2end/async_end2end_test.cc338
-rw-r--r--test/cpp/interop/stress_interop_client.cc40
-rw-r--r--test/cpp/interop/stress_interop_client.h8
-rw-r--r--test/cpp/interop/stress_test.cc13
-rw-r--r--test/cpp/qps/client.h12
-rw-r--r--test/cpp/qps/driver.cc3
-rwxr-xr-xtest/cpp/qps/gen_build_yaml.py10
-rw-r--r--test/cpp/util/metrics_server.cc45
-rw-r--r--test/cpp/util/metrics_server.h40
11 files changed, 665 insertions, 202 deletions
diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden
new file mode 100644
index 0000000000..ef3d1aaa51
--- /dev/null
+++ b/test/cpp/codegen/compiler_test_golden
@@ -0,0 +1,294 @@
+// Generated by the gRPC protobuf plugin.
+// If you make any local change, they will be lost.
+// source: src/proto/grpc/testing/compiler_test.proto
+// Original file comments:
+// Copyright 2016, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+// * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+// * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+// * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+// File detached comment 1
+//
+// File detached comment 2
+//
+// File leading comment 1
+#ifndef GRPC_src_2fproto_2fgrpc_2ftesting_2fcompiler_5ftest_2eproto__INCLUDED
+#define GRPC_src_2fproto_2fgrpc_2ftesting_2fcompiler_5ftest_2eproto__INCLUDED
+
+#include "src/proto/grpc/testing/compiler_test.pb.h"
+
+#include <grpc++/impl/codegen/async_stream.h>
+#include <grpc++/impl/codegen/async_unary_call.h>
+#include <grpc++/impl/codegen/proto_utils.h>
+#include <grpc++/impl/codegen/rpc_method.h>
+#include <grpc++/impl/codegen/service_type.h>
+#include <grpc++/impl/codegen/status.h>
+#include <grpc++/impl/codegen/stub_options.h>
+#include <grpc++/impl/codegen/sync_stream.h>
+
+namespace grpc {
+class CompletionQueue;
+class Channel;
+class RpcService;
+class ServerCompletionQueue;
+class ServerContext;
+} // namespace grpc
+
+namespace grpc {
+namespace testing {
+
+// ServiceA detached comment 1
+//
+// ServiceA detached comment 2
+//
+// ServiceA leading comment 1
+class ServiceA GRPC_FINAL {
+ public:
+ class StubInterface {
+ public:
+ virtual ~StubInterface() {}
+ // MethodA1 leading comment 1
+ virtual ::grpc::Status MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) = 0;
+ std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> AsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+ return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq));
+ }
+ // MethodA1 trailing comment 1
+ // MethodA2 detached leading comment 1
+ //
+ // Method A2 leading comment 1
+ // Method A2 leading comment 2
+ std::unique_ptr< ::grpc::ClientWriterInterface< ::grpc::testing::Request>> MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response) {
+ return std::unique_ptr< ::grpc::ClientWriterInterface< ::grpc::testing::Request>>(MethodA2Raw(context, response));
+ }
+ std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>> AsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) {
+ return std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>>(AsyncMethodA2Raw(context, response, cq, tag));
+ }
+ // MethodA2 trailing comment 1
+ private:
+ virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* AsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
+ virtual ::grpc::ClientWriterInterface< ::grpc::testing::Request>* MethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response) = 0;
+ virtual ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>* AsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) = 0;
+ };
+ class Stub GRPC_FINAL : public StubInterface {
+ public:
+ Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel);
+ ::grpc::Status MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) GRPC_OVERRIDE;
+ std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> AsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+ return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq));
+ }
+ std::unique_ptr< ::grpc::ClientWriter< ::grpc::testing::Request>> MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response) {
+ return std::unique_ptr< ::grpc::ClientWriter< ::grpc::testing::Request>>(MethodA2Raw(context, response));
+ }
+ std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>> AsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) {
+ return std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>>(AsyncMethodA2Raw(context, response, cq, tag));
+ }
+
+ private:
+ std::shared_ptr< ::grpc::ChannelInterface> channel_;
+ ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) GRPC_OVERRIDE;
+ ::grpc::ClientWriter< ::grpc::testing::Request>* MethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response) GRPC_OVERRIDE;
+ ::grpc::ClientAsyncWriter< ::grpc::testing::Request>* AsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) GRPC_OVERRIDE;
+ const ::grpc::RpcMethod rpcmethod_MethodA1_;
+ const ::grpc::RpcMethod rpcmethod_MethodA2_;
+ };
+ static std::unique_ptr<Stub> NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions());
+
+ class Service : public ::grpc::Service {
+ public:
+ Service();
+ virtual ~Service();
+ // MethodA1 leading comment 1
+ virtual ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response);
+ // MethodA1 trailing comment 1
+ // MethodA2 detached leading comment 1
+ //
+ // Method A2 leading comment 1
+ // Method A2 leading comment 2
+ virtual ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response);
+ // MethodA2 trailing comment 1
+ };
+ template <class BaseClass>
+ class WithAsyncMethod_MethodA1 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithAsyncMethod_MethodA1() {
+ ::grpc::Service::MarkMethodAsync(0);
+ }
+ ~WithAsyncMethod_MethodA1() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable synchronous version of this method
+ ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ void RequestMethodA1(::grpc::ServerContext* context, ::grpc::testing::Request* request, ::grpc::ServerAsyncResponseWriter< ::grpc::testing::Response>* response, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) {
+ ::grpc::Service::RequestAsyncUnary(0, context, request, response, new_call_cq, notification_cq, tag);
+ }
+ };
+ template <class BaseClass>
+ class WithAsyncMethod_MethodA2 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithAsyncMethod_MethodA2() {
+ ::grpc::Service::MarkMethodAsync(1);
+ }
+ ~WithAsyncMethod_MethodA2() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable synchronous version of this method
+ ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ void RequestMethodA2(::grpc::ServerContext* context, ::grpc::ServerAsyncReader< ::grpc::testing::Response, ::grpc::testing::Request>* reader, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) {
+ ::grpc::Service::RequestAsyncClientStreaming(1, context, reader, new_call_cq, notification_cq, tag);
+ }
+ };
+ typedef WithAsyncMethod_MethodA1<WithAsyncMethod_MethodA2<Service > > AsyncService;
+ template <class BaseClass>
+ class WithGenericMethod_MethodA1 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithGenericMethod_MethodA1() {
+ ::grpc::Service::MarkMethodGeneric(0);
+ }
+ ~WithGenericMethod_MethodA1() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable synchronous version of this method
+ ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ };
+ template <class BaseClass>
+ class WithGenericMethod_MethodA2 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithGenericMethod_MethodA2() {
+ ::grpc::Service::MarkMethodGeneric(1);
+ }
+ ~WithGenericMethod_MethodA2() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable synchronous version of this method
+ ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ };
+};
+
+// ServiceB leading comment 1
+class ServiceB GRPC_FINAL {
+ public:
+ class StubInterface {
+ public:
+ virtual ~StubInterface() {}
+ // MethodB1 leading comment 1
+ virtual ::grpc::Status MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) = 0;
+ std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> AsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+ return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq));
+ }
+ // MethodB1 trailing comment 1
+ private:
+ virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
+ };
+ class Stub GRPC_FINAL : public StubInterface {
+ public:
+ Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel);
+ ::grpc::Status MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) GRPC_OVERRIDE;
+ std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> AsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+ return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq));
+ }
+
+ private:
+ std::shared_ptr< ::grpc::ChannelInterface> channel_;
+ ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) GRPC_OVERRIDE;
+ const ::grpc::RpcMethod rpcmethod_MethodB1_;
+ };
+ static std::unique_ptr<Stub> NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions());
+
+ class Service : public ::grpc::Service {
+ public:
+ Service();
+ virtual ~Service();
+ // MethodB1 leading comment 1
+ virtual ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response);
+ // MethodB1 trailing comment 1
+ };
+ template <class BaseClass>
+ class WithAsyncMethod_MethodB1 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithAsyncMethod_MethodB1() {
+ ::grpc::Service::MarkMethodAsync(0);
+ }
+ ~WithAsyncMethod_MethodB1() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable synchronous version of this method
+ ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ void RequestMethodB1(::grpc::ServerContext* context, ::grpc::testing::Request* request, ::grpc::ServerAsyncResponseWriter< ::grpc::testing::Response>* response, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) {
+ ::grpc::Service::RequestAsyncUnary(0, context, request, response, new_call_cq, notification_cq, tag);
+ }
+ };
+ typedef WithAsyncMethod_MethodB1<Service > AsyncService;
+ template <class BaseClass>
+ class WithGenericMethod_MethodB1 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithGenericMethod_MethodB1() {
+ ::grpc::Service::MarkMethodGeneric(0);
+ }
+ ~WithGenericMethod_MethodB1() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable synchronous version of this method
+ ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ };
+};
+// ServiceB trailing comment 1
+
+} // namespace testing
+} // namespace grpc
+
+
+#endif // GRPC_src_2fproto_2fgrpc_2ftesting_2fcompiler_5ftest_2eproto__INCLUDED
diff --git a/test/cpp/codegen/golden_file_test.cc b/test/cpp/codegen/golden_file_test.cc
new file mode 100644
index 0000000000..ec08d08de6
--- /dev/null
+++ b/test/cpp/codegen/golden_file_test.cc
@@ -0,0 +1,64 @@
+/*
+ *
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <fstream>
+#include <sstream>
+
+#include <gtest/gtest.h>
+
+// These paths rely on the fact that we run our tests under grpc/
+const char kGeneratedFilePath[] =
+ "gens/src/proto/grpc/testing/compiler_test.grpc.pb.h";
+const char kGoldenFilePath[] = "test/cpp/codegen/compiler_test_golden";
+
+TEST(GoldenFileTest, TestGeneratedFile) {
+ std::ifstream generated(kGeneratedFilePath);
+ std::ifstream golden(kGoldenFilePath);
+
+ ASSERT_TRUE(generated.good());
+ ASSERT_TRUE(golden.good());
+
+ std::ostringstream gen_oss;
+ std::ostringstream gold_oss;
+ gen_oss << generated.rdbuf();
+ gold_oss << golden.rdbuf();
+ EXPECT_EQ(gold_oss.str(), gen_oss.str());
+
+ generated.close();
+ golden.close();
+}
+
+int main(int argc, char **argv) {
+ ::testing::InitGoogleTest(&argc, argv);
+ return RUN_ALL_TESTS();
+}
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index 7e4d6046d6..0232a9fa31 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -51,6 +51,7 @@
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
#include "test/cpp/util/string_ref_helper.h"
+#include "test/cpp/util/test_credentials_provider.h"
#ifdef GPR_POSIX_SOCKET
#include "src/core/lib/iomgr/ev_posix.h"
@@ -58,6 +59,7 @@
using grpc::testing::EchoRequest;
using grpc::testing::EchoResponse;
+using grpc::testing::kTlsCredentialsType;
using std::chrono::system_clock;
GPR_TLS_DECL(g_is_async_end2end_test);
@@ -197,20 +199,37 @@ class Verifier {
bool spin_;
};
-class AsyncEnd2endTest : public ::testing::TestWithParam<bool> {
+class TestScenario {
+ public:
+ TestScenario(bool non_block, const grpc::string& creds_type,
+ const grpc::string& content)
+ : disable_blocking(non_block),
+ credentials_type(creds_type),
+ message_content(content) {}
+ void Log() const {
+ gpr_log(GPR_INFO,
+ "Scenario: disable_blocking %d, credentials %s, message size %d",
+ disable_blocking, credentials_type.c_str(), message_content.size());
+ }
+ bool disable_blocking;
+ const grpc::string credentials_type;
+ const grpc::string message_content;
+};
+
+class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
protected:
- AsyncEnd2endTest() {}
+ AsyncEnd2endTest() { GetParam().Log(); }
void SetUp() GRPC_OVERRIDE {
- poll_overrider_.reset(new PollingOverrider(!GetParam()));
+ poll_overrider_.reset(new PollingOverrider(!GetParam().disable_blocking));
int port = grpc_pick_unused_port_or_die();
server_address_ << "localhost:" << port;
// Setup server
ServerBuilder builder;
- builder.AddListeningPort(server_address_.str(),
- grpc::InsecureServerCredentials());
+ auto server_creds = GetServerCredentials(GetParam().credentials_type);
+ builder.AddListeningPort(server_address_.str(), server_creds);
builder.RegisterService(&service_);
cq_ = builder.AddCompletionQueue();
server_ = builder.BuildAndStart();
@@ -230,8 +249,11 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<bool> {
}
void ResetStub() {
+ ChannelArguments args;
+ auto channel_creds =
+ GetChannelCredentials(GetParam().credentials_type, &args);
std::shared_ptr<Channel> channel =
- CreateChannel(server_address_.str(), InsecureChannelCredentials());
+ CreateCustomChannel(server_address_.str(), channel_creds, args);
stub_ = grpc::testing::EchoTestService::NewStub(channel);
}
@@ -247,22 +269,23 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<bool> {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, cq_.get()));
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
response_writer.Finish(send_response, Status::OK, tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
-
response_reader->Finish(&recv_response, &recv_status, tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
+ .Expect(4, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
@@ -302,7 +325,7 @@ TEST_P(AsyncEnd2endTest, AsyncNextRpc) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, cq_.get()));
@@ -310,23 +333,22 @@ TEST_P(AsyncEnd2endTest, AsyncNextRpc) {
std::chrono::system_clock::now());
std::chrono::system_clock::time_point time_limit(
std::chrono::system_clock::now() + std::chrono::seconds(10));
- Verifier(GetParam()).Verify(cq_.get(), time_now);
- Verifier(GetParam()).Verify(cq_.get(), time_now);
+ Verifier(GetParam().disable_blocking).Verify(cq_.get(), time_now);
+ Verifier(GetParam().disable_blocking).Verify(cq_.get(), time_now);
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get(), time_limit);
+ Verifier(GetParam().disable_blocking)
+ .Expect(2, true)
+ .Verify(cq_.get(), time_limit);
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
response_writer.Finish(send_response, Status::OK, tag(3));
- Verifier(GetParam())
- .Expect(3, true)
- .Verify(cq_.get(), std::chrono::system_clock::time_point::max());
-
response_reader->Finish(&recv_response, &recv_status, tag(4));
- Verifier(GetParam())
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
.Expect(4, true)
.Verify(cq_.get(), std::chrono::system_clock::time_point::max());
@@ -347,41 +369,48 @@ TEST_P(AsyncEnd2endTest, SimpleClientStreaming) {
ServerContext srv_ctx;
ServerAsyncReader<EchoResponse, EchoRequest> srv_stream(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncWriter<EchoRequest>> cli_stream(
stub_->AsyncRequestStream(&cli_ctx, &recv_response, cq_.get(), tag(1)));
service_.RequestRequestStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
- Verifier(GetParam()).Expect(2, true).Expect(1, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(2, true)
+ .Expect(1, true)
+ .Verify(cq_.get());
cli_stream->Write(send_request, tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
-
srv_stream.Read(&recv_request, tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
+ .Expect(4, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
cli_stream->Write(send_request, tag(5));
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
-
srv_stream.Read(&recv_request, tag(6));
- Verifier(GetParam()).Expect(6, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(5, true)
+ .Expect(6, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
cli_stream->WritesDone(tag(7));
- Verifier(GetParam()).Expect(7, true).Verify(cq_.get());
-
srv_stream.Read(&recv_request, tag(8));
- Verifier(GetParam()).Expect(8, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(7, true)
+ .Expect(8, false)
+ .Verify(cq_.get());
send_response.set_message(recv_request.message());
srv_stream.Finish(send_response, Status::OK, tag(9));
- Verifier(GetParam()).Expect(9, true).Verify(cq_.get());
-
cli_stream->Finish(&recv_status, tag(10));
- Verifier(GetParam()).Expect(10, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(9, true)
+ .Expect(10, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
@@ -400,39 +429,45 @@ TEST_P(AsyncEnd2endTest, SimpleServerStreaming) {
ServerContext srv_ctx;
ServerAsyncWriter<EchoResponse> srv_stream(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncReader<EchoResponse>> cli_stream(
stub_->AsyncResponseStream(&cli_ctx, send_request, cq_.get(), tag(1)));
service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream,
cq_.get(), cq_.get(), tag(2));
- Verifier(GetParam()).Expect(1, true).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(1, true)
+ .Expect(2, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
srv_stream.Write(send_response, tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
-
cli_stream->Read(&recv_response, tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
+ .Expect(4, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
srv_stream.Write(send_response, tag(5));
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
-
cli_stream->Read(&recv_response, tag(6));
- Verifier(GetParam()).Expect(6, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(5, true)
+ .Expect(6, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
srv_stream.Finish(Status::OK, tag(7));
- Verifier(GetParam()).Expect(7, true).Verify(cq_.get());
-
cli_stream->Read(&recv_response, tag(8));
- Verifier(GetParam()).Expect(8, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(7, true)
+ .Expect(8, false)
+ .Verify(cq_.get());
cli_stream->Finish(&recv_status, tag(9));
- Verifier(GetParam()).Expect(9, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(9, true).Verify(cq_.get());
EXPECT_TRUE(recv_status.ok());
}
@@ -450,41 +485,48 @@ TEST_P(AsyncEnd2endTest, SimpleBidiStreaming) {
ServerContext srv_ctx;
ServerAsyncReaderWriter<EchoResponse, EchoRequest> srv_stream(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse>>
cli_stream(stub_->AsyncBidiStream(&cli_ctx, cq_.get(), tag(1)));
service_.RequestBidiStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
- Verifier(GetParam()).Expect(1, true).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(1, true)
+ .Expect(2, true)
+ .Verify(cq_.get());
cli_stream->Write(send_request, tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
-
srv_stream.Read(&recv_request, tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
+ .Expect(4, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
srv_stream.Write(send_response, tag(5));
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
-
cli_stream->Read(&recv_response, tag(6));
- Verifier(GetParam()).Expect(6, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(5, true)
+ .Expect(6, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
cli_stream->WritesDone(tag(7));
- Verifier(GetParam()).Expect(7, true).Verify(cq_.get());
-
srv_stream.Read(&recv_request, tag(8));
- Verifier(GetParam()).Expect(8, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(7, true)
+ .Expect(8, false)
+ .Verify(cq_.get());
srv_stream.Finish(Status::OK, tag(9));
- Verifier(GetParam()).Expect(9, true).Verify(cq_.get());
-
cli_stream->Finish(&recv_status, tag(10));
- Verifier(GetParam()).Expect(10, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(9, true)
+ .Expect(10, true)
+ .Verify(cq_.get());
EXPECT_TRUE(recv_status.ok());
}
@@ -503,7 +545,7 @@ TEST_P(AsyncEnd2endTest, ClientInitialMetadataRpc) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::pair<grpc::string, grpc::string> meta1("key1", "val1");
std::pair<grpc::string, grpc::string> meta2("key2", "val2");
std::pair<grpc::string, grpc::string> meta3("g.r.d-bin", "xyz");
@@ -516,7 +558,7 @@ TEST_P(AsyncEnd2endTest, ClientInitialMetadataRpc) {
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
auto client_initial_metadata = srv_ctx.client_metadata();
EXPECT_EQ(meta1.second,
@@ -529,11 +571,11 @@ TEST_P(AsyncEnd2endTest, ClientInitialMetadataRpc) {
send_response.set_message(recv_request.message());
response_writer.Finish(send_response, Status::OK, tag(3));
-
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
-
response_reader->Finish(&recv_response, &recv_status, tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
+ .Expect(4, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
@@ -552,7 +594,7 @@ TEST_P(AsyncEnd2endTest, ServerInitialMetadataRpc) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::pair<grpc::string, grpc::string> meta1("key1", "val1");
std::pair<grpc::string, grpc::string> meta2("key2", "val2");
@@ -561,15 +603,15 @@ TEST_P(AsyncEnd2endTest, ServerInitialMetadataRpc) {
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
srv_ctx.AddInitialMetadata(meta1.first, meta1.second);
srv_ctx.AddInitialMetadata(meta2.first, meta2.second);
response_writer.SendInitialMetadata(tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(3, true).Verify(cq_.get());
response_reader->ReadInitialMetadata(tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
auto server_initial_metadata = cli_ctx.GetServerInitialMetadata();
EXPECT_EQ(meta1.second,
ToString(server_initial_metadata.find(meta1.first)->second));
@@ -579,10 +621,11 @@ TEST_P(AsyncEnd2endTest, ServerInitialMetadataRpc) {
send_response.set_message(recv_request.message());
response_writer.Finish(send_response, Status::OK, tag(5));
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
-
response_reader->Finish(&recv_response, &recv_status, tag(6));
- Verifier(GetParam()).Expect(6, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(5, true)
+ .Expect(6, true)
+ .Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
@@ -601,7 +644,7 @@ TEST_P(AsyncEnd2endTest, ServerTrailingMetadataRpc) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::pair<grpc::string, grpc::string> meta1("key1", "val1");
std::pair<grpc::string, grpc::string> meta2("key2", "val2");
@@ -610,20 +653,22 @@ TEST_P(AsyncEnd2endTest, ServerTrailingMetadataRpc) {
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
response_writer.SendInitialMetadata(tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(3, true).Verify(cq_.get());
send_response.set_message(recv_request.message());
srv_ctx.AddTrailingMetadata(meta1.first, meta1.second);
srv_ctx.AddTrailingMetadata(meta2.first, meta2.second);
response_writer.Finish(send_response, Status::OK, tag(4));
+ response_reader->Finish(&recv_response, &recv_status, tag(5));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(4, true)
+ .Expect(5, true)
+ .Verify(cq_.get());
- response_reader->Finish(&recv_response, &recv_status, tag(5));
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata();
@@ -647,7 +692,7 @@ TEST_P(AsyncEnd2endTest, MetadataRpc) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::pair<grpc::string, grpc::string> meta1("key1", "val1");
std::pair<grpc::string, grpc::string> meta2(
"key2-bin",
@@ -671,7 +716,7 @@ TEST_P(AsyncEnd2endTest, MetadataRpc) {
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
auto client_initial_metadata = srv_ctx.client_metadata();
EXPECT_EQ(meta1.second,
@@ -683,9 +728,9 @@ TEST_P(AsyncEnd2endTest, MetadataRpc) {
srv_ctx.AddInitialMetadata(meta3.first, meta3.second);
srv_ctx.AddInitialMetadata(meta4.first, meta4.second);
response_writer.SendInitialMetadata(tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(3, true).Verify(cq_.get());
response_reader->ReadInitialMetadata(tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
auto server_initial_metadata = cli_ctx.GetServerInitialMetadata();
EXPECT_EQ(meta3.second,
ToString(server_initial_metadata.find(meta3.first)->second));
@@ -697,11 +742,13 @@ TEST_P(AsyncEnd2endTest, MetadataRpc) {
srv_ctx.AddTrailingMetadata(meta5.first, meta5.second);
srv_ctx.AddTrailingMetadata(meta6.first, meta6.second);
response_writer.Finish(send_response, Status::OK, tag(5));
+ response_reader->Finish(&recv_response, &recv_status, tag(6));
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(5, true)
+ .Expect(6, true)
+ .Verify(cq_.get());
- response_reader->Finish(&recv_response, &recv_status, tag(6));
- Verifier(GetParam()).Expect(6, true).Verify(cq_.get());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata();
@@ -726,7 +773,7 @@ TEST_P(AsyncEnd2endTest, ServerCheckCancellation) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, cq_.get()));
@@ -734,15 +781,15 @@ TEST_P(AsyncEnd2endTest, ServerCheckCancellation) {
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
cli_ctx.TryCancel();
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(5, true).Verify(cq_.get());
EXPECT_TRUE(srv_ctx.IsCancelled());
response_reader->Finish(&recv_response, &recv_status, tag(4));
- Verifier(GetParam()).Expect(4, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(4, false).Verify(cq_.get());
EXPECT_EQ(StatusCode::CANCELLED, recv_status.error_code());
}
@@ -761,7 +808,7 @@ TEST_P(AsyncEnd2endTest, ServerCheckDone) {
ServerContext srv_ctx;
grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, cq_.get()));
@@ -769,25 +816,29 @@ TEST_P(AsyncEnd2endTest, ServerCheckDone) {
service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(),
cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
response_writer.Finish(send_response, Status::OK, tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
- Verifier(GetParam()).Expect(5, true).Verify(cq_.get());
- EXPECT_FALSE(srv_ctx.IsCancelled());
-
response_reader->Finish(&recv_response, &recv_status, tag(4));
- Verifier(GetParam()).Expect(4, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(3, true)
+ .Expect(4, true)
+ .Expect(5, true)
+ .Verify(cq_.get());
+ EXPECT_FALSE(srv_ctx.IsCancelled());
EXPECT_EQ(send_response.message(), recv_response.message());
EXPECT_TRUE(recv_status.ok());
}
TEST_P(AsyncEnd2endTest, UnimplementedRpc) {
+ ChannelArguments args;
+ auto channel_creds =
+ GetChannelCredentials(GetParam().credentials_type, &args);
std::shared_ptr<Channel> channel =
- CreateChannel(server_address_.str(), InsecureChannelCredentials());
+ CreateCustomChannel(server_address_.str(), channel_creds, args);
std::unique_ptr<grpc::testing::UnimplementedService::Stub> stub;
stub = grpc::testing::UnimplementedService::NewStub(channel);
EchoRequest send_request;
@@ -795,12 +846,12 @@ TEST_P(AsyncEnd2endTest, UnimplementedRpc) {
Status recv_status;
ClientContext cli_ctx;
- send_request.set_message("Hello");
+ send_request.set_message(GetParam().message_content);
std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
stub->AsyncUnimplemented(&cli_ctx, send_request, cq_.get()));
response_reader->Finish(&recv_response, &recv_status, tag(4));
- Verifier(GetParam()).Expect(4, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(4, false).Verify(cq_.get());
EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code());
EXPECT_EQ("", recv_status.error_message());
@@ -847,23 +898,25 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Initiate the 'RequestStream' call on client
std::unique_ptr<ClientAsyncWriter<EchoRequest>> cli_stream(
stub_->AsyncRequestStream(&cli_ctx, &recv_response, cq_.get(), tag(1)));
- Verifier(GetParam()).Expect(1, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(1, true).Verify(cq_.get());
// On the server, request to be notified of 'RequestStream' calls
// and receive the 'RequestStream' call just made by the client
srv_ctx.AsyncNotifyWhenDone(tag(11));
service_.RequestRequestStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
// Client sends 3 messages (tags 3, 4 and 5)
for (int tag_idx = 3; tag_idx <= 5; tag_idx++) {
send_request.set_message("Ping " + std::to_string(tag_idx));
cli_stream->Write(send_request, tag(tag_idx));
- Verifier(GetParam()).Expect(tag_idx, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking)
+ .Expect(tag_idx, true)
+ .Verify(cq_.get());
}
cli_stream->WritesDone(tag(6));
- Verifier(GetParam()).Expect(6, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(6, true).Verify(cq_.get());
bool expected_server_cq_result = true;
bool ignore_cq_result = false;
@@ -871,7 +924,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
srv_ctx.TryCancel();
- Verifier(GetParam()).Expect(11, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(11, true).Verify(cq_.get());
EXPECT_TRUE(srv_ctx.IsCancelled());
// Since cancellation is done before server reads any results, we know
@@ -881,7 +934,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
std::thread* server_try_cancel_thd = NULL;
- auto verif = Verifier(GetParam());
+ auto verif = Verifier(GetParam().disable_blocking);
if (server_try_cancel == CANCEL_DURING_PROCESSING) {
server_try_cancel_thd =
@@ -939,13 +992,13 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Server sends the final message and cancelled status (but the RPC is
// already cancelled at this point. So we expect the operation to fail)
srv_stream.Finish(send_response, Status::CANCELLED, tag(9));
- Verifier(GetParam()).Expect(9, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(9, false).Verify(cq_.get());
// Client will see the cancellation
cli_stream->Finish(&recv_status, tag(10));
// TODO(sreek): The expectation here should be true. This is a bug (github
// issue #4972)
- Verifier(GetParam()).Expect(10, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(10, false).Verify(cq_.get());
EXPECT_FALSE(recv_status.ok());
EXPECT_EQ(::grpc::StatusCode::CANCELLED, recv_status.error_code());
}
@@ -979,13 +1032,13 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Initiate the 'ResponseStream' call on the client
std::unique_ptr<ClientAsyncReader<EchoResponse>> cli_stream(
stub_->AsyncResponseStream(&cli_ctx, send_request, cq_.get(), tag(1)));
- Verifier(GetParam()).Expect(1, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(1, true).Verify(cq_.get());
// On the server, request to be notified of 'ResponseStream' calls and
// receive the call just made by the client
srv_ctx.AsyncNotifyWhenDone(tag(11));
service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream,
cq_.get(), cq_.get(), tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
EXPECT_EQ(send_request.message(), recv_request.message());
bool expected_cq_result = true;
@@ -994,7 +1047,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
srv_ctx.TryCancel();
- Verifier(GetParam()).Expect(11, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(11, true).Verify(cq_.get());
EXPECT_TRUE(srv_ctx.IsCancelled());
// We know for sure that all cq results will be false from this point
@@ -1004,7 +1057,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
std::thread* server_try_cancel_thd = NULL;
- auto verif = Verifier(GetParam());
+ auto verif = Verifier(GetParam().disable_blocking);
if (server_try_cancel == CANCEL_DURING_PROCESSING) {
server_try_cancel_thd =
@@ -1064,7 +1117,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Client attemts to read the three messages from the server
for (int tag_idx = 6; tag_idx <= 8; tag_idx++) {
cli_stream->Read(&recv_response, tag(tag_idx));
- Verifier(GetParam())
+ Verifier(GetParam().disable_blocking)
.Expect(tag_idx, expected_cq_result)
.Verify(cq_.get(), ignore_cq_result);
}
@@ -1075,11 +1128,11 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Server finishes the stream (but the RPC is already cancelled)
srv_stream.Finish(Status::CANCELLED, tag(9));
- Verifier(GetParam()).Expect(9, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(9, false).Verify(cq_.get());
// Client will see the cancellation
cli_stream->Finish(&recv_status, tag(10));
- Verifier(GetParam()).Expect(10, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(10, true).Verify(cq_.get());
EXPECT_FALSE(recv_status.ok());
EXPECT_EQ(::grpc::StatusCode::CANCELLED, recv_status.error_code());
}
@@ -1114,19 +1167,19 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Initiate the call from the client side
std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse>>
cli_stream(stub_->AsyncBidiStream(&cli_ctx, cq_.get(), tag(1)));
- Verifier(GetParam()).Expect(1, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(1, true).Verify(cq_.get());
// On the server, request to be notified of the 'BidiStream' call and
// receive the call just made by the client
srv_ctx.AsyncNotifyWhenDone(tag(11));
service_.RequestBidiStream(&srv_ctx, &srv_stream, cq_.get(), cq_.get(),
tag(2));
- Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(2, true).Verify(cq_.get());
// Client sends the first and the only message
send_request.set_message("Ping");
cli_stream->Write(send_request, tag(3));
- Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(3, true).Verify(cq_.get());
bool expected_cq_result = true;
bool ignore_cq_result = false;
@@ -1134,7 +1187,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
srv_ctx.TryCancel();
- Verifier(GetParam()).Expect(11, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(11, true).Verify(cq_.get());
EXPECT_TRUE(srv_ctx.IsCancelled());
// We know for sure that all cq results will be false from this point
@@ -1144,7 +1197,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
std::thread* server_try_cancel_thd = NULL;
- auto verif = Verifier(GetParam());
+ auto verif = Verifier(GetParam().disable_blocking);
if (server_try_cancel == CANCEL_DURING_PROCESSING) {
server_try_cancel_thd =
@@ -1244,10 +1297,10 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// know that cq results are supposed to return false on server.
srv_stream.Finish(Status::CANCELLED, tag(9));
- Verifier(GetParam()).Expect(9, false).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(9, false).Verify(cq_.get());
cli_stream->Finish(&recv_status, tag(10));
- Verifier(GetParam()).Expect(10, true).Verify(cq_.get());
+ Verifier(GetParam().disable_blocking).Expect(10, true).Verify(cq_.get());
EXPECT_FALSE(recv_status.ok());
EXPECT_EQ(grpc::StatusCode::CANCELLED, recv_status.error_code());
}
@@ -1289,11 +1342,48 @@ TEST_P(AsyncEnd2endServerTryCancelTest, ServerBidiStreamingTryCancelAfter) {
TestBidiStreamingServerCancel(CANCEL_AFTER_PROCESSING);
}
+std::vector<TestScenario> CreateTestScenarios(bool test_disable_blocking,
+ bool test_secure,
+ int test_big_limit) {
+ std::vector<TestScenario> scenarios;
+ std::vector<grpc::string> credentials_types;
+ std::vector<grpc::string> messages;
+
+ credentials_types.push_back(kInsecureCredentialsType);
+ auto sec_list = GetSecureCredentialsTypeList();
+ for (auto sec = sec_list.begin(); sec != sec_list.end(); sec++) {
+ credentials_types.push_back(*sec);
+ }
+
+ messages.push_back("Hello");
+ for (int sz = 1; sz < test_big_limit; sz *= 2) {
+ grpc::string big_msg;
+ for (int i = 0; i < sz * 1024; i++) {
+ char c = 'a' + (i % 26);
+ big_msg += c;
+ }
+ messages.push_back(big_msg);
+ }
+
+ for (auto cred = credentials_types.begin(); cred != credentials_types.end();
+ ++cred) {
+ for (auto msg = messages.begin(); msg != messages.end(); msg++) {
+ scenarios.push_back(TestScenario(false, *cred, *msg));
+ if (test_disable_blocking) {
+ scenarios.push_back(TestScenario(true, *cred, *msg));
+ }
+ }
+ }
+ return scenarios;
+}
+
INSTANTIATE_TEST_CASE_P(AsyncEnd2end, AsyncEnd2endTest,
- ::testing::Values(false, true));
+ ::testing::ValuesIn(CreateTestScenarios(true, true,
+ 1024)));
INSTANTIATE_TEST_CASE_P(AsyncEnd2endServerTryCancel,
AsyncEnd2endServerTryCancelTest,
- ::testing::Values(false));
+ ::testing::ValuesIn(CreateTestScenarios(false, false,
+ 0)));
} // namespace
} // namespace testing
diff --git a/test/cpp/interop/stress_interop_client.cc b/test/cpp/interop/stress_interop_client.cc
index 04671fb935..f287a5aa3b 100644
--- a/test/cpp/interop/stress_interop_client.cc
+++ b/test/cpp/interop/stress_interop_client.cc
@@ -84,49 +84,37 @@ StressTestInteropClient::StressTestInteropClient(
int test_id, const grpc::string& server_address,
std::shared_ptr<Channel> channel,
const WeightedRandomTestSelector& test_selector, long test_duration_secs,
- long sleep_duration_ms, long metrics_collection_interval_secs)
+ long sleep_duration_ms)
: test_id_(test_id),
server_address_(server_address),
channel_(channel),
interop_client_(new InteropClient(channel, false)),
test_selector_(test_selector),
test_duration_secs_(test_duration_secs),
- sleep_duration_ms_(sleep_duration_ms),
- metrics_collection_interval_secs_(metrics_collection_interval_secs) {}
+ sleep_duration_ms_(sleep_duration_ms) {}
-void StressTestInteropClient::MainLoop(std::shared_ptr<Gauge> qps_gauge) {
+void StressTestInteropClient::MainLoop(std::shared_ptr<QpsGauge> qps_gauge) {
gpr_log(GPR_INFO, "Running test %d. ServerAddr: %s", test_id_,
server_address_.c_str());
- gpr_timespec test_end_time =
- gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
- gpr_time_from_seconds(test_duration_secs_, GPR_TIMESPAN));
+ gpr_timespec test_end_time;
+ if (test_duration_secs_ < 0) {
+ test_end_time = gpr_inf_future(GPR_CLOCK_REALTIME);
+ } else {
+ test_end_time =
+ gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+ gpr_time_from_seconds(test_duration_secs_, GPR_TIMESPAN));
+ }
- gpr_timespec current_time = gpr_now(GPR_CLOCK_REALTIME);
- gpr_timespec next_stat_collection_time = current_time;
- gpr_timespec collection_interval =
- gpr_time_from_seconds(metrics_collection_interval_secs_, GPR_TIMESPAN);
- long num_calls_per_interval = 0;
+ qps_gauge->Reset();
- while (test_duration_secs_ < 0 ||
- gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), test_end_time) < 0) {
+ while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), test_end_time) < 0) {
// Select the test case to execute based on the weights and execute it
TestCaseType test_case = test_selector_.GetNextTest();
gpr_log(GPR_DEBUG, "%d - Executing the test case %d", test_id_, test_case);
RunTest(test_case);
- num_calls_per_interval++;
-
- // See if its time to collect stats yet
- current_time = gpr_now(GPR_CLOCK_REALTIME);
- if (gpr_time_cmp(next_stat_collection_time, current_time) < 0) {
- qps_gauge->Set(num_calls_per_interval /
- metrics_collection_interval_secs_);
-
- num_calls_per_interval = 0;
- next_stat_collection_time =
- gpr_time_add(current_time, collection_interval);
- }
+ qps_gauge->Incr();
// Sleep between successive calls if needed
if (sleep_duration_ms_ > 0) {
diff --git a/test/cpp/interop/stress_interop_client.h b/test/cpp/interop/stress_interop_client.h
index 6fd303d6b7..cb0cd98821 100644
--- a/test/cpp/interop/stress_interop_client.h
+++ b/test/cpp/interop/stress_interop_client.h
@@ -87,12 +87,11 @@ class StressTestInteropClient {
StressTestInteropClient(int test_id, const grpc::string& server_address,
std::shared_ptr<Channel> channel,
const WeightedRandomTestSelector& test_selector,
- long test_duration_secs, long sleep_duration_ms,
- long metrics_collection_interval_secs);
+ long test_duration_secs, long sleep_duration_ms);
// The main function. Use this as the thread entry point.
- // qps_gauge is the Gauge to record the requests per second metric
- void MainLoop(std::shared_ptr<Gauge> qps_gauge);
+ // qps_gauge is the QpsGauge to record the requests per second metric
+ void MainLoop(std::shared_ptr<QpsGauge> qps_gauge);
private:
void RunTest(TestCaseType test_case);
@@ -104,7 +103,6 @@ class StressTestInteropClient {
const WeightedRandomTestSelector& test_selector_;
long test_duration_secs_;
long sleep_duration_ms_;
- long metrics_collection_interval_secs_;
};
} // namespace testing
diff --git a/test/cpp/interop/stress_test.cc b/test/cpp/interop/stress_test.cc
index 38caf31b76..d9e3fd25c5 100644
--- a/test/cpp/interop/stress_test.cc
+++ b/test/cpp/interop/stress_test.cc
@@ -56,9 +56,6 @@ extern void gpr_default_log(gpr_log_func_args* args);
DEFINE_int32(metrics_port, 8081, "The metrics server port.");
-DEFINE_int32(metrics_collection_interval_secs, 5,
- "How often (in seconds) should metrics be recorded.");
-
DEFINE_int32(sleep_duration_ms, 0,
"The duration (in millisec) between two"
" consecutive test calls (per server) issued by the server.");
@@ -275,19 +272,19 @@ int main(int argc, char** argv) {
stub_idx++) {
StressTestInteropClient* client = new StressTestInteropClient(
++thread_idx, *it, channel, test_selector, FLAGS_test_duration_secs,
- FLAGS_sleep_duration_ms, FLAGS_metrics_collection_interval_secs);
+ FLAGS_sleep_duration_ms);
- bool is_already_created;
- // Gauge name
+ bool is_already_created = false;
+ // QpsGauge name
std::snprintf(buffer, sizeof(buffer),
"/stress_test/server_%d/channel_%d/stub_%d/qps",
server_idx, channel_idx, stub_idx);
test_threads.emplace_back(grpc::thread(
&StressTestInteropClient::MainLoop, client,
- metrics_service.CreateGauge(buffer, &is_already_created)));
+ metrics_service.CreateQpsGauge(buffer, &is_already_created)));
- // The Gauge should not have been already created
+ // The QpsGauge should not have been already created
GPR_ASSERT(!is_already_created);
}
}
diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h
index 5a9027a4a2..175529f01b 100644
--- a/test/cpp/qps/client.h
+++ b/test/cpp/qps/client.h
@@ -39,6 +39,7 @@
#include <vector>
#include <grpc++/support/byte_buffer.h>
+#include <grpc++/support/channel_arguments.h>
#include <grpc++/support/slice.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
@@ -280,7 +281,7 @@ class ClientImpl : public Client {
create_stub_(create_stub) {
for (int i = 0; i < config.client_channels(); i++) {
channels_[i].init(config.server_targets(i % config.server_targets_size()),
- config, create_stub_);
+ config, create_stub_, i);
}
ClientRequestCreator<RequestType> create_req(&request_,
@@ -303,14 +304,17 @@ class ClientImpl : public Client {
}
void init(const grpc::string& target, const ClientConfig& config,
std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)>
- create_stub) {
+ create_stub,
+ int shard) {
// We have to use a 2-phase init like this with a default
// constructor followed by an initializer function to make
// old compilers happy with using this in std::vector
+ ChannelArguments args;
+ args.SetInt("shard_to_ensure_no_subchannel_merges", shard);
channel_ = CreateTestChannel(
target, config.security_params().server_host_override(),
- config.has_security_params(),
- !config.security_params().use_test_ca());
+ config.has_security_params(), !config.security_params().use_test_ca(),
+ std::shared_ptr<CallCredentials>(), args);
stub_ = create_stub(channel_);
}
Channel* get_channel() { return channel_.get(); }
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index 2583ceb819..04b2b453f9 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -83,6 +83,7 @@ static std::unordered_map<string, std::deque<int>> get_hosts_and_cores(
auto stub = WorkerService::NewStub(
CreateChannel(*it, InsecureChannelCredentials()));
grpc::ClientContext ctx;
+ ctx.set_fail_fast(false);
CoreRequest dummy;
CoreResponse cores;
grpc::Status s = stub->CoreCount(&ctx, dummy, &cores);
@@ -166,6 +167,7 @@ namespace runsc {
static ClientContext* AllocContext(list<ClientContext>* contexts) {
contexts->emplace_back();
auto context = &contexts->back();
+ context->set_fail_fast(false);
return context;
}
@@ -435,6 +437,7 @@ void RunQuit() {
CreateChannel(workers[i], InsecureChannelCredentials()));
Void dummy;
grpc::ClientContext ctx;
+ ctx.set_fail_fast(false);
GPR_ASSERT(stub->QuitWorker(&ctx, dummy, &dummy).ok());
}
}
diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py
index 9d6bf2ab73..6b3329b046 100755
--- a/test/cpp/qps/gen_build_yaml.py
+++ b/test/cpp/qps/gen_build_yaml.py
@@ -43,12 +43,16 @@ sys.path.append(run_tests_root)
import performance.scenario_config as scenario_config
+def _scenario_json_string(scenario_json):
+ return json.dumps(scenario_config.remove_nonproto_fields(scenario_json))
+
print yaml.dump({
'tests': [
{
'name': 'json_run_localhost',
- 'shortname': 'json_run_localhost:%s' % js['name'],
- 'args': ['--scenario_json', pipes.quote(json.dumps(js))],
+ 'shortname': 'json_run_localhost:%s' % scenario_json['name'],
+ 'args': ['--scenario_json',
+ pipes.quote(_scenario_json_string(scenario_json))],
'ci_platforms': ['linux', 'mac', 'posix', 'windows'],
'platforms': ['linux', 'mac', 'posix', 'windows'],
'flaky': False,
@@ -58,6 +62,6 @@ print yaml.dump({
'cpu_cost': 1000.0,
'exclude_configs': []
}
- for js in scenario_config.CXXLanguage().scenarios()
+ for scenario_json in scenario_config.CXXLanguage().scenarios()
]
})
diff --git a/test/cpp/util/metrics_server.cc b/test/cpp/util/metrics_server.cc
index d9b44a6a92..cc6b39b753 100644
--- a/test/cpp/util/metrics_server.cc
+++ b/test/cpp/util/metrics_server.cc
@@ -42,16 +42,26 @@
namespace grpc {
namespace testing {
-Gauge::Gauge(long initial_val) : val_(initial_val) {}
+QpsGauge::QpsGauge()
+ : start_time_(gpr_now(GPR_CLOCK_REALTIME)), num_queries_(0) {}
-void Gauge::Set(long new_val) {
- std::lock_guard<std::mutex> lock(val_mu_);
- val_ = new_val;
+void QpsGauge::Reset() {
+ std::lock_guard<std::mutex> lock(num_queries_mu_);
+ num_queries_ = 0;
+ start_time_ = gpr_now(GPR_CLOCK_REALTIME);
}
-long Gauge::Get() {
- std::lock_guard<std::mutex> lock(val_mu_);
- return val_;
+void QpsGauge::Incr() {
+ std::lock_guard<std::mutex> lock(num_queries_mu_);
+ num_queries_++;
+}
+
+long QpsGauge::Get() {
+ std::lock_guard<std::mutex> lock(num_queries_mu_);
+ gpr_timespec time_diff =
+ gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME), start_time_);
+ long duration_secs = time_diff.tv_sec > 0 ? time_diff.tv_sec : 1;
+ return num_queries_ / duration_secs;
}
grpc::Status MetricsServiceImpl::GetAllGauges(
@@ -60,7 +70,7 @@ grpc::Status MetricsServiceImpl::GetAllGauges(
gpr_log(GPR_DEBUG, "GetAllGauges called");
std::lock_guard<std::mutex> lock(mu_);
- for (auto it = gauges_.begin(); it != gauges_.end(); it++) {
+ for (auto it = qps_gauges_.begin(); it != qps_gauges_.end(); it++) {
GaugeResponse resp;
resp.set_name(it->first); // Gauge name
resp.set_long_value(it->second->Get()); // Gauge value
@@ -75,8 +85,8 @@ grpc::Status MetricsServiceImpl::GetGauge(ServerContext* context,
GaugeResponse* response) {
std::lock_guard<std::mutex> lock(mu_);
- const auto it = gauges_.find(request->name());
- if (it != gauges_.end()) {
+ const auto it = qps_gauges_.find(request->name());
+ if (it != qps_gauges_.end()) {
response->set_name(it->first);
response->set_long_value(it->second->Get());
}
@@ -84,16 +94,17 @@ grpc::Status MetricsServiceImpl::GetGauge(ServerContext* context,
return Status::OK;
}
-std::shared_ptr<Gauge> MetricsServiceImpl::CreateGauge(const grpc::string& name,
- bool* already_present) {
+std::shared_ptr<QpsGauge> MetricsServiceImpl::CreateQpsGauge(
+ const grpc::string& name, bool* already_present) {
std::lock_guard<std::mutex> lock(mu_);
- std::shared_ptr<Gauge> gauge(new Gauge(0));
- const auto p = gauges_.emplace(name, gauge);
+ std::shared_ptr<QpsGauge> qps_gauge(new QpsGauge());
+ const auto p = qps_gauges_.emplace(name, qps_gauge);
- // p.first is an iterator pointing to <name, shared_ptr<Gauge>> pair. p.second
- // is a boolean which is set to 'true' if the Gauge is inserted in the guages_
- // map and 'false' if it is already present in the map
+ // p.first is an iterator pointing to <name, shared_ptr<QpsGauge>> pair.
+ // p.second is a boolean which is set to 'true' if the QpsGauge is
+ // successfully inserted in the guages_ map and 'false' if it is already
+ // present in the map
*already_present = !p.second;
return p.first->second;
}
diff --git a/test/cpp/util/metrics_server.h b/test/cpp/util/metrics_server.h
index ce05e0be64..aa9bfed23d 100644
--- a/test/cpp/util/metrics_server.h
+++ b/test/cpp/util/metrics_server.h
@@ -48,10 +48,13 @@
* Example:
* MetricsServiceImpl metricsImpl;
* ..
- * // Create Gauge(s). Note: Gauges can be created even after calling
+ * // Create QpsGauge(s). Note: QpsGauges can be created even after calling
* // 'StartServer'.
- * Gauge gauge1 = metricsImpl.CreateGauge("foo",is_present);
- * // gauge1 can now be used anywhere in the program to set values.
+ * QpsGauge qps_gauge1 = metricsImpl.CreateQpsGauge("foo", is_present);
+ * // qps_gauge1 can now be used anywhere in the program by first making a
+ * // one-time call qps_gauge1.Reset() and then calling qps_gauge1.Incr()
+ * // every time to increment a query counter
+ *
* ...
* // Create the metrics server
* std::unique_ptr<grpc::Server> server = metricsImpl.StartServer(port);
@@ -60,17 +63,24 @@
namespace grpc {
namespace testing {
-// TODO(sreek): Add support for other types of Gauges like Double, String in
-// future
-class Gauge {
+class QpsGauge {
public:
- Gauge(long initial_val);
- void Set(long new_val);
+ QpsGauge();
+
+ // Initialize the internal timer and reset the query count to 0
+ void Reset();
+
+ // Increment the query count by 1
+ void Incr();
+
+ // Return the current qps (i.e query count divided by the time since this
+ // QpsGauge object created (or Reset() was called))
long Get();
private:
- long val_;
- std::mutex val_mu_;
+ gpr_timespec start_time_;
+ long num_queries_;
+ std::mutex num_queries_mu_;
};
class MetricsServiceImpl GRPC_FINAL : public MetricsService::Service {
@@ -81,17 +91,17 @@ class MetricsServiceImpl GRPC_FINAL : public MetricsService::Service {
grpc::Status GetGauge(ServerContext* context, const GaugeRequest* request,
GaugeResponse* response) GRPC_OVERRIDE;
- // Create a Gauge with name 'name'. is_present is set to true if the Gauge
+ // Create a QpsGauge with name 'name'. is_present is set to true if the Gauge
// is already present in the map.
- // NOTE: CreateGauge can be called anytime (i.e before or after calling
+ // NOTE: CreateQpsGauge can be called anytime (i.e before or after calling
// StartServer).
- std::shared_ptr<Gauge> CreateGauge(const grpc::string& name,
- bool* already_present);
+ std::shared_ptr<QpsGauge> CreateQpsGauge(const grpc::string& name,
+ bool* already_present);
std::unique_ptr<grpc::Server> StartServer(int port);
private:
- std::map<string, std::shared_ptr<Gauge>> gauges_;
+ std::map<string, std::shared_ptr<QpsGauge>> qps_gauges_;
std::mutex mu_;
};
generated by cgit v1.2.3 (git 2.39.1) at 2024-10-03 16:09:55 +0000