aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp/interop
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp/interop')
-rw-r--r--test/cpp/interop/client.cc114
-rw-r--r--test/cpp/interop/client_helper.cc19
-rw-r--r--test/cpp/interop/interop_client.cc667
-rw-r--r--test/cpp/interop/interop_client.h70
-rw-r--r--test/cpp/interop/interop_server.cc (renamed from test/cpp/interop/server_main.cc)143
-rw-r--r--test/cpp/interop/metrics_client.cc18
-rw-r--r--test/cpp/interop/rnd.datbin524288 -> 0 bytes
-rw-r--r--test/cpp/interop/server_helper.cc4
-rw-r--r--test/cpp/interop/server_helper.h1
-rw-r--r--test/cpp/interop/stress_interop_client.cc69
-rw-r--r--test/cpp/interop/stress_interop_client.h44
-rw-r--r--test/cpp/interop/stress_test.cc21
12 files changed, 793 insertions, 377 deletions
diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc
index 9af6a88044..e8ae6ee572 100644
--- a/test/cpp/interop/client.cc
+++ b/test/cpp/interop/client.cc
@@ -40,7 +40,9 @@
#include <grpc++/client_context.h>
#include <grpc/grpc.h>
#include <grpc/support/log.h>
+#include <grpc/support/useful.h>
+#include "src/core/lib/support/string.h"
#include "test/cpp/interop/client_helper.h"
#include "test/cpp/interop/interop_client.h"
#include "test/cpp/util/test_config.h"
@@ -52,35 +54,44 @@ DEFINE_string(server_host, "127.0.0.1", "Server host to connect to");
DEFINE_string(server_host_override, "foo.test.google.fr",
"Override the server host which is sent in HTTP header");
DEFINE_string(test_case, "large_unary",
- "Configure different test cases. Valid options are: "
- "empty_unary : empty (zero bytes) request and response; "
- "large_unary : single request and (large) response; "
- "large_compressed_unary : single request and compressed (large) "
- "response; "
- "client_streaming : request streaming with single response; "
- "server_streaming : single request with response streaming; "
+ "Configure different test cases. Valid options are:\n\n"
+ "all : all test cases;\n"
+ "cancel_after_begin : cancel stream after starting it;\n"
+ "cancel_after_first_response: cancel on first response;\n"
+ "client_compressed_streaming : compressed request streaming with "
+ "client_compressed_unary : single compressed request;\n"
+ "client_streaming : request streaming with single response;\n"
+ "compute_engine_creds: large_unary with compute engine auth;\n"
+ "custom_metadata: server will echo custom metadata;\n"
+ "empty_stream : bi-di stream with no request/response;\n"
+ "empty_unary : empty (zero bytes) request and response;\n"
+ "half_duplex : half-duplex streaming;\n"
+ "jwt_token_creds: large_unary with JWT token auth;\n"
+ "large_unary : single request and (large) response;\n"
+ "oauth2_auth_token: raw oauth2 access token auth;\n"
+ "per_rpc_creds: raw oauth2 access token on a single rpc;\n"
+ "ping_pong : full-duplex streaming;\n"
+ "response streaming;\n"
"server_compressed_streaming : single request with compressed "
- "response streaming; "
- "slow_consumer : single request with response; "
- " streaming with slow client consumer; "
- "half_duplex : half-duplex streaming; "
- "ping_pong : full-duplex streaming; "
- "cancel_after_begin : cancel stream after starting it; "
- "cancel_after_first_response: cancel on first response; "
- "timeout_on_sleeping_server: deadline exceeds on stream; "
- "empty_stream : bi-di stream with no request/response; "
- "compute_engine_creds: large_unary with compute engine auth; "
- "jwt_token_creds: large_unary with JWT token auth; "
- "oauth2_auth_token: raw oauth2 access token auth; "
- "per_rpc_creds: raw oauth2 access token on a single rpc; "
- "status_code_and_message: verify status code & message; "
- "custom_metadata: server will echo custom metadata;"
- "all : all of above.");
+ "server_compressed_unary : single compressed response;\n"
+ "server_streaming : single request with response streaming;\n"
+ "slow_consumer : single request with response streaming with "
+ "slow client consumer;\n"
+ "status_code_and_message: verify status code & message;\n"
+ "timeout_on_sleeping_server: deadline exceeds on stream;\n");
DEFINE_string(default_service_account, "",
"Email of GCE default service account");
DEFINE_string(service_account_key_file, "",
"Path to service account json key file.");
DEFINE_string(oauth_scope, "", "Scope for OAuth tokens.");
+DEFINE_bool(do_not_abort_on_transient_failures, false,
+ "If set to 'true', abort() is not called in case of transient "
+ "failures (i.e failures that are temporary and will likely go away "
+ "on retrying; like a temporary connection failure) and an error "
+ "message is printed instead. Note that this flag just controls "
+ "whether abort() is called or not. It does not control whether the "
+ "test is retried in case of transient failures (and currently the "
+ "interop tests are not retried even if this flag is set to true)");
using grpc::testing::CreateChannelForTestCase;
using grpc::testing::GetServiceAccountJsonKey;
@@ -89,20 +100,25 @@ int main(int argc, char** argv) {
grpc::testing::InitTest(&argc, &argv, true);
gpr_log(GPR_INFO, "Testing these cases: %s", FLAGS_test_case.c_str());
int ret = 0;
- grpc::testing::InteropClient client(
- CreateChannelForTestCase(FLAGS_test_case));
+ grpc::testing::InteropClient client(CreateChannelForTestCase(FLAGS_test_case),
+ true,
+ FLAGS_do_not_abort_on_transient_failures);
if (FLAGS_test_case == "empty_unary") {
client.DoEmpty();
} else if (FLAGS_test_case == "large_unary") {
client.DoLargeUnary();
- } else if (FLAGS_test_case == "large_compressed_unary") {
- client.DoLargeCompressedUnary();
+ } else if (FLAGS_test_case == "server_compressed_unary") {
+ client.DoServerCompressedUnary();
+ } else if (FLAGS_test_case == "client_compressed_unary") {
+ client.DoClientCompressedUnary();
} else if (FLAGS_test_case == "client_streaming") {
client.DoRequestStreaming();
} else if (FLAGS_test_case == "server_streaming") {
client.DoResponseStreaming();
} else if (FLAGS_test_case == "server_compressed_streaming") {
- client.DoResponseCompressedStreaming();
+ client.DoServerCompressedStreaming();
+ } else if (FLAGS_test_case == "client_compressed_streaming") {
+ client.DoClientCompressedStreaming();
} else if (FLAGS_test_case == "slow_consumer") {
client.DoResponseStreamingWithSlowConsumer();
} else if (FLAGS_test_case == "half_duplex") {
@@ -135,9 +151,12 @@ int main(int argc, char** argv) {
} else if (FLAGS_test_case == "all") {
client.DoEmpty();
client.DoLargeUnary();
+ client.DoClientCompressedUnary();
+ client.DoServerCompressedUnary();
client.DoRequestStreaming();
client.DoResponseStreaming();
- client.DoResponseCompressedStreaming();
+ client.DoClientCompressedStreaming();
+ client.DoServerCompressedStreaming();
client.DoHalfDuplex();
client.DoPingPong();
client.DoCancelAfterBegin();
@@ -156,14 +175,35 @@ int main(int argc, char** argv) {
}
// compute_engine_creds only runs in GCE.
} else {
- gpr_log(
- GPR_ERROR,
- "Unsupported test case %s. Valid options are all|empty_unary|"
- "large_unary|large_compressed_unary|client_streaming|server_streaming|"
- "server_compressed_streaming|half_duplex|ping_pong|cancel_after_begin|"
- "cancel_after_first_response|timeout_on_sleeping_server|empty_stream|"
- "compute_engine_creds|jwt_token_creds|oauth2_auth_token|per_rpc_creds",
- "status_code_and_message|custom_metadata", FLAGS_test_case.c_str());
+ const char* testcases[] = {"all",
+ "cancel_after_begin",
+ "cancel_after_first_response",
+ "client_compressed_streaming",
+ "client_compressed_unary",
+ "client_streaming",
+ "compute_engine_creds",
+ "custom_metadata",
+ "empty_stream",
+ "empty_unary",
+ "half_duplex",
+ "jwt_token_creds",
+ "large_unary",
+ "oauth2_auth_token",
+ "oauth2_auth_token",
+ "per_rpc_creds",
+ "per_rpc_creds",
+ "ping_pong",
+ "server_compressed_streaming",
+ "server_compressed_unary",
+ "server_streaming",
+ "status_code_and_message",
+ "timeout_on_sleeping_server"};
+ char* joined_testcases =
+ gpr_strjoin_sep(testcases, GPR_ARRAY_SIZE(testcases), "\n", NULL);
+
+ gpr_log(GPR_ERROR, "Unsupported test case %s. Valid options are\n%s",
+ FLAGS_test_case.c_str(), joined_testcases);
+ gpr_free(joined_testcases);
ret = 1;
}
diff --git a/test/cpp/interop/client_helper.cc b/test/cpp/interop/client_helper.cc
index c8b1e505c1..c171969e14 100644
--- a/test/cpp/interop/client_helper.cc
+++ b/test/cpp/interop/client_helper.cc
@@ -97,30 +97,25 @@ std::shared_ptr<Channel> CreateChannelForTestCase(
snprintf(host_port, host_port_buf_size, "%s:%d", FLAGS_server_host.c_str(),
FLAGS_server_port);
+ std::shared_ptr<CallCredentials> creds;
if (test_case == "compute_engine_creds") {
- std::shared_ptr<CallCredentials> creds;
GPR_ASSERT(FLAGS_use_tls);
creds = GoogleComputeEngineCredentials();
- return CreateTestChannel(host_port, FLAGS_server_host_override,
- FLAGS_use_tls, !FLAGS_use_test_ca, creds);
+ GPR_ASSERT(creds);
} else if (test_case == "jwt_token_creds") {
- std::shared_ptr<CallCredentials> creds;
GPR_ASSERT(FLAGS_use_tls);
grpc::string json_key = GetServiceAccountJsonKey();
std::chrono::seconds token_lifetime = std::chrono::hours(1);
creds =
ServiceAccountJWTAccessCredentials(json_key, token_lifetime.count());
- return CreateTestChannel(host_port, FLAGS_server_host_override,
- FLAGS_use_tls, !FLAGS_use_test_ca, creds);
+ GPR_ASSERT(creds);
} else if (test_case == "oauth2_auth_token") {
grpc::string raw_token = GetOauth2AccessToken();
- std::shared_ptr<CallCredentials> creds = AccessTokenCredentials(raw_token);
- return CreateTestChannel(host_port, FLAGS_server_host_override,
- FLAGS_use_tls, !FLAGS_use_test_ca, creds);
- } else {
- return CreateTestChannel(host_port, FLAGS_server_host_override,
- FLAGS_use_tls, !FLAGS_use_test_ca);
+ creds = AccessTokenCredentials(raw_token);
+ GPR_ASSERT(creds);
}
+ return CreateTestChannel(host_port, FLAGS_server_host_override, FLAGS_use_tls,
+ !FLAGS_use_test_ca, creds);
}
} // namespace testing
diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc
index 22293d211f..8861bc1163 100644
--- a/test/cpp/interop/interop_client.cc
+++ b/test/cpp/interop/interop_client.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -31,10 +31,8 @@
*
*/
-#include "test/cpp/interop/interop_client.h"
-
#include <unistd.h>
-
+#include <cinttypes>
#include <fstream>
#include <memory>
@@ -51,12 +49,11 @@
#include "src/proto/grpc/testing/messages.grpc.pb.h"
#include "src/proto/grpc/testing/test.grpc.pb.h"
#include "test/cpp/interop/client_helper.h"
+#include "test/cpp/interop/interop_client.h"
namespace grpc {
namespace testing {
-static const char* kRandomFile = "test/cpp/interop/rnd.dat";
-
namespace {
// The same value is defined by the Java client.
const std::vector<int> request_stream_sizes = {27182, 8, 1828, 45904};
@@ -67,35 +64,26 @@ const int kReceiveDelayMilliSeconds = 20;
const int kLargeRequestSize = 271828;
const int kLargeResponseSize = 314159;
-CompressionType GetInteropCompressionTypeFromCompressionAlgorithm(
- grpc_compression_algorithm algorithm) {
- switch (algorithm) {
- case GRPC_COMPRESS_NONE:
- return CompressionType::NONE;
- case GRPC_COMPRESS_GZIP:
- return CompressionType::GZIP;
- case GRPC_COMPRESS_DEFLATE:
- return CompressionType::DEFLATE;
- default:
- GPR_ASSERT(false);
- }
-}
-
void NoopChecks(const InteropClientContextInspector& inspector,
const SimpleRequest* request, const SimpleResponse* response) {}
-void CompressionChecks(const InteropClientContextInspector& inspector,
- const SimpleRequest* request,
- const SimpleResponse* response) {
- GPR_ASSERT(request->response_compression() ==
- GetInteropCompressionTypeFromCompressionAlgorithm(
- inspector.GetCallCompressionAlgorithm()));
- if (request->response_compression() == NONE) {
- GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
- } else if (request->response_type() == PayloadType::COMPRESSABLE) {
- // requested compression and compressable response => results should always
- // be compressed.
+void UnaryCompressionChecks(const InteropClientContextInspector& inspector,
+ const SimpleRequest* request,
+ const SimpleResponse* response) {
+ const grpc_compression_algorithm received_compression =
+ inspector.GetCallCompressionAlgorithm();
+ if (request->response_compressed().value()) {
+ if (received_compression == GRPC_COMPRESS_NONE) {
+ // Requested some compression, got NONE. This is an error.
+ gpr_log(GPR_ERROR,
+ "Failure: Requested compression but got uncompressed response "
+ "from server.");
+ abort();
+ }
GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
+ } else {
+ // Didn't request compression -> make sure the response is uncompressed
+ GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
}
}
} // namespace
@@ -134,23 +122,43 @@ void InteropClient::Reset(std::shared_ptr<Channel> channel) {
serviceStub_.Reset(channel);
}
-InteropClient::InteropClient(std::shared_ptr<Channel> channel)
- : serviceStub_(channel, true) {}
-
InteropClient::InteropClient(std::shared_ptr<Channel> channel,
- bool new_stub_every_test_case)
- : serviceStub_(channel, new_stub_every_test_case) {}
+ bool new_stub_every_test_case,
+ bool do_not_abort_on_transient_failures)
+ : serviceStub_(channel, new_stub_every_test_case),
+ do_not_abort_on_transient_failures_(do_not_abort_on_transient_failures) {}
-void InteropClient::AssertOkOrPrintErrorStatus(const Status& s) {
+bool InteropClient::AssertStatusOk(const Status& s) {
if (s.ok()) {
- return;
+ return true;
+ }
+
+ // Note: At this point, s.error_code is definitely not StatusCode::OK (we
+ // already checked for s.ok() above). So, the following will call abort()
+ // (unless s.error_code() corresponds to a transient failure and
+ // 'do_not_abort_on_transient_failures' is true)
+ return AssertStatusCode(s, StatusCode::OK);
+}
+
+bool InteropClient::AssertStatusCode(const Status& s,
+ StatusCode expected_code) {
+ if (s.error_code() == expected_code) {
+ return true;
}
- gpr_log(GPR_ERROR, "Error status code: %d, message: %s", s.error_code(),
- s.error_message().c_str());
- GPR_ASSERT(0);
+
+ gpr_log(GPR_ERROR, "Error status code: %d (expected: %d), message: %s",
+ s.error_code(), expected_code, s.error_message().c_str());
+
+ // In case of transient transient/retryable failures (like a broken
+ // connection) we may or may not abort (see TransientFailureOrAbort())
+ if (s.error_code() == grpc::StatusCode::UNAVAILABLE) {
+ return TransientFailureOrAbort();
+ }
+
+ abort();
}
-void InteropClient::DoEmpty() {
+bool InteropClient::DoEmpty() {
gpr_log(GPR_DEBUG, "Sending an empty rpc...");
Empty request = Empty::default_instance();
@@ -158,54 +166,50 @@ void InteropClient::DoEmpty() {
ClientContext context;
Status s = serviceStub_.Get()->EmptyCall(&context, request, &response);
- AssertOkOrPrintErrorStatus(s);
+
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
gpr_log(GPR_DEBUG, "Empty rpc done.");
+ return true;
}
-void InteropClient::PerformLargeUnary(SimpleRequest* request,
+bool InteropClient::PerformLargeUnary(SimpleRequest* request,
SimpleResponse* response) {
- PerformLargeUnary(request, response, NoopChecks);
+ return PerformLargeUnary(request, response, NoopChecks);
}
-void InteropClient::PerformLargeUnary(SimpleRequest* request,
+bool InteropClient::PerformLargeUnary(SimpleRequest* request,
SimpleResponse* response,
CheckerFn custom_checks_fn) {
ClientContext context;
InteropClientContextInspector inspector(context);
- // If the request doesn't already specify the response type, default to
- // COMPRESSABLE.
request->set_response_size(kLargeResponseSize);
grpc::string payload(kLargeRequestSize, '\0');
request->mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
+ if (request->has_expect_compressed()) {
+ if (request->expect_compressed().value()) {
+ context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
+ } else {
+ context.set_compression_algorithm(GRPC_COMPRESS_NONE);
+ }
+ }
Status s = serviceStub_.Get()->UnaryCall(&context, *request, response);
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
custom_checks_fn(inspector, request, response);
// Payload related checks.
- if (request->response_type() != PayloadType::RANDOM) {
- GPR_ASSERT(response->payload().type() == request->response_type());
- }
- switch (response->payload().type()) {
- case PayloadType::COMPRESSABLE:
- GPR_ASSERT(response->payload().body() ==
- grpc::string(kLargeResponseSize, '\0'));
- break;
- case PayloadType::UNCOMPRESSABLE: {
- std::ifstream rnd_file(kRandomFile);
- GPR_ASSERT(rnd_file.good());
- for (int i = 0; i < kLargeResponseSize; i++) {
- GPR_ASSERT(response->payload().body()[i] == (char)rnd_file.get());
- }
- } break;
- default:
- GPR_ASSERT(false);
- }
+ GPR_ASSERT(response->payload().body() ==
+ grpc::string(kLargeResponseSize, '\0'));
+ return true;
}
-void InteropClient::DoComputeEngineCreds(
+bool InteropClient::DoComputeEngineCreds(
const grpc::string& default_service_account,
const grpc::string& oauth_scope) {
gpr_log(GPR_DEBUG,
@@ -214,8 +218,11 @@ void InteropClient::DoComputeEngineCreds(
SimpleResponse response;
request.set_fill_username(true);
request.set_fill_oauth_scope(true);
- request.set_response_type(PayloadType::COMPRESSABLE);
- PerformLargeUnary(&request, &response);
+
+ if (!PerformLargeUnary(&request, &response)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
gpr_log(GPR_DEBUG, "Got oauth_scope %s", response.oauth_scope().c_str());
GPR_ASSERT(!response.username().empty());
@@ -224,9 +231,10 @@ void InteropClient::DoComputeEngineCreds(
const char* oauth_scope_str = response.oauth_scope().c_str();
GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
gpr_log(GPR_DEBUG, "Large unary with compute engine creds done.");
+ return true;
}
-void InteropClient::DoOauth2AuthToken(const grpc::string& username,
+bool InteropClient::DoOauth2AuthToken(const grpc::string& username,
const grpc::string& oauth_scope) {
gpr_log(GPR_DEBUG,
"Sending a unary rpc with raw oauth2 access token credentials ...");
@@ -239,16 +247,20 @@ void InteropClient::DoOauth2AuthToken(const grpc::string& username,
Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
GPR_ASSERT(!response.username().empty());
GPR_ASSERT(!response.oauth_scope().empty());
GPR_ASSERT(username == response.username());
const char* oauth_scope_str = response.oauth_scope().c_str();
GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
gpr_log(GPR_DEBUG, "Unary with oauth2 access token credentials done.");
+ return true;
}
-void InteropClient::DoPerRpcCreds(const grpc::string& json_key) {
+bool InteropClient::DoPerRpcCreds(const grpc::string& json_key) {
gpr_log(GPR_DEBUG, "Sending a unary rpc with per-rpc JWT access token ...");
SimpleRequest request;
SimpleResponse response;
@@ -263,58 +275,127 @@ void InteropClient::DoPerRpcCreds(const grpc::string& json_key) {
Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
GPR_ASSERT(!response.username().empty());
GPR_ASSERT(json_key.find(response.username()) != grpc::string::npos);
gpr_log(GPR_DEBUG, "Unary with per-rpc JWT access token done.");
+ return true;
}
-void InteropClient::DoJwtTokenCreds(const grpc::string& username) {
+bool InteropClient::DoJwtTokenCreds(const grpc::string& username) {
gpr_log(GPR_DEBUG,
"Sending a large unary rpc with JWT token credentials ...");
SimpleRequest request;
SimpleResponse response;
request.set_fill_username(true);
- request.set_response_type(PayloadType::COMPRESSABLE);
- PerformLargeUnary(&request, &response);
+
+ if (!PerformLargeUnary(&request, &response)) {
+ return false;
+ }
+
GPR_ASSERT(!response.username().empty());
GPR_ASSERT(username.find(response.username()) != grpc::string::npos);
gpr_log(GPR_DEBUG, "Large unary with JWT token creds done.");
+ return true;
}
-void InteropClient::DoLargeUnary() {
+bool InteropClient::DoLargeUnary() {
gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
SimpleRequest request;
SimpleResponse response;
- request.set_response_type(PayloadType::COMPRESSABLE);
- PerformLargeUnary(&request, &response);
+ if (!PerformLargeUnary(&request, &response)) {
+ return false;
+ }
gpr_log(GPR_DEBUG, "Large unary done.");
+ return true;
}
-void InteropClient::DoLargeCompressedUnary() {
- const CompressionType compression_types[] = {NONE, GZIP, DEFLATE};
- const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE, RANDOM};
- for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
- for (size_t j = 0; j < GPR_ARRAY_SIZE(compression_types); j++) {
- char* log_suffix;
- gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
- CompressionType_Name(compression_types[j]).c_str(),
- PayloadType_Name(payload_types[i]).c_str());
-
- gpr_log(GPR_DEBUG, "Sending a large compressed unary rpc %s.",
- log_suffix);
- SimpleRequest request;
- SimpleResponse response;
- request.set_response_type(payload_types[i]);
- request.set_response_compression(compression_types[j]);
- PerformLargeUnary(&request, &response, CompressionChecks);
- gpr_log(GPR_DEBUG, "Large compressed unary done %s.", log_suffix);
+bool InteropClient::DoClientCompressedUnary() {
+ // Probing for compression-checks support.
+ ClientContext probe_context;
+ SimpleRequest probe_req;
+ SimpleResponse probe_res;
+
+ probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
+ probe_req.mutable_expect_compressed()->set_value(true); // lies!
+
+ probe_req.set_response_size(kLargeResponseSize);
+ probe_req.mutable_payload()->set_body(grpc::string(kLargeRequestSize, '\0'));
+
+ gpr_log(GPR_DEBUG, "Sending probe for compressed unary request.");
+ const Status s =
+ serviceStub_.Get()->UnaryCall(&probe_context, probe_req, &probe_res);
+ if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
+ // The server isn't able to evaluate incoming compression, making the rest
+ // of this test moot.
+ gpr_log(GPR_DEBUG, "Compressed unary request probe failed");
+ return false;
+ }
+ gpr_log(GPR_DEBUG, "Compressed unary request probe succeeded. Proceeding.");
+
+ const std::vector<bool> compressions = {true, false};
+ for (size_t i = 0; i < compressions.size(); i++) {
+ char* log_suffix;
+ gpr_asprintf(&log_suffix, "(compression=%s)",
+ compressions[i] ? "true" : "false");
+
+ gpr_log(GPR_DEBUG, "Sending compressed unary request %s.", log_suffix);
+ SimpleRequest request;
+ SimpleResponse response;
+ request.mutable_expect_compressed()->set_value(compressions[i]);
+ if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
+ gpr_log(GPR_ERROR, "Compressed unary request failed %s", log_suffix);
gpr_free(log_suffix);
+ return false;
}
+
+ gpr_log(GPR_DEBUG, "Compressed unary request failed %s", log_suffix);
+ gpr_free(log_suffix);
}
+
+ return true;
}
-void InteropClient::DoRequestStreaming() {
+bool InteropClient::DoServerCompressedUnary() {
+ const std::vector<bool> compressions = {true, false};
+ for (size_t i = 0; i < compressions.size(); i++) {
+ char* log_suffix;
+ gpr_asprintf(&log_suffix, "(compression=%s)",
+ compressions[i] ? "true" : "false");
+
+ gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.",
+ log_suffix);
+ SimpleRequest request;
+ SimpleResponse response;
+ request.mutable_response_compressed()->set_value(compressions[i]);
+
+ if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
+ gpr_log(GPR_ERROR, "Request for compressed unary failed %s", log_suffix);
+ gpr_free(log_suffix);
+ return false;
+ }
+
+ gpr_log(GPR_DEBUG, "Request for compressed unary failed %s", log_suffix);
+ gpr_free(log_suffix);
+ }
+
+ return true;
+}
+
+// Either abort() (unless do_not_abort_on_transient_failures_ is true) or return
+// false
+bool InteropClient::TransientFailureOrAbort() {
+ if (do_not_abort_on_transient_failures_) {
+ return false;
+ }
+
+ abort();
+}
+
+bool InteropClient::DoRequestStreaming() {
gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
ClientContext context;
@@ -325,22 +406,28 @@ void InteropClient::DoRequestStreaming() {
serviceStub_.Get()->StreamingInputCall(&context, &response));
int aggregated_payload_size = 0;
- for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
+ for (size_t i = 0; i < request_stream_sizes.size(); ++i) {
Payload* payload = request.mutable_payload();
payload->set_body(grpc::string(request_stream_sizes[i], '\0'));
- GPR_ASSERT(stream->Write(request));
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "DoRequestStreaming(): stream->Write() failed");
+ return TransientFailureOrAbort();
+ }
aggregated_payload_size += request_stream_sizes[i];
}
- stream->WritesDone();
+ GPR_ASSERT(stream->WritesDone());
+
Status s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size);
- AssertOkOrPrintErrorStatus(s);
- gpr_log(GPR_DEBUG, "Request streaming done.");
+ return true;
}
-void InteropClient::DoResponseStreaming() {
- gpr_log(GPR_DEBUG, "Receiving response steaming rpc ...");
+bool InteropClient::DoResponseStreaming() {
+ gpr_log(GPR_DEBUG, "Receiving response streaming rpc ...");
ClientContext context;
StreamingOutputCallRequest request;
@@ -358,92 +445,153 @@ void InteropClient::DoResponseStreaming() {
grpc::string(response_stream_sizes[i], '\0'));
++i;
}
- GPR_ASSERT(response_stream_sizes.size() == i);
+
+ if (i < response_stream_sizes.size()) {
+ // stream->Read() failed before reading all the expected messages. This is
+ // most likely due to connection failure.
+ gpr_log(GPR_ERROR,
+ "DoResponseStreaming(): Read fewer streams (%d) than "
+ "response_stream_sizes.size() (%" PRIuPTR ")",
+ i, response_stream_sizes.size());
+ return TransientFailureOrAbort();
+ }
+
Status s = stream->Finish();
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "Response streaming done.");
+ return true;
}
-void InteropClient::DoResponseCompressedStreaming() {
- const CompressionType compression_types[] = {NONE, GZIP, DEFLATE};
- const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE, RANDOM};
- for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
- for (size_t j = 0; j < GPR_ARRAY_SIZE(compression_types); j++) {
- ClientContext context;
- InteropClientContextInspector inspector(context);
- StreamingOutputCallRequest request;
-
- char* log_suffix;
- gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
- CompressionType_Name(compression_types[j]).c_str(),
- PayloadType_Name(payload_types[i]).c_str());
-
- gpr_log(GPR_DEBUG, "Receiving response steaming rpc %s.", log_suffix);
-
- request.set_response_type(payload_types[i]);
- request.set_response_compression(compression_types[j]);
-
- for (size_t k = 0; k < response_stream_sizes.size(); ++k) {
- ResponseParameters* response_parameter =
- request.add_response_parameters();
- response_parameter->set_size(response_stream_sizes[k]);
- }
- StreamingOutputCallResponse response;
-
- std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
- serviceStub_.Get()->StreamingOutputCall(&context, request));
-
- size_t k = 0;
- while (stream->Read(&response)) {
- // Payload related checks.
- if (request.response_type() != PayloadType::RANDOM) {
- GPR_ASSERT(response.payload().type() == request.response_type());
- }
- switch (response.payload().type()) {
- case PayloadType::COMPRESSABLE:
- GPR_ASSERT(response.payload().body() ==
- grpc::string(response_stream_sizes[k], '\0'));
- break;
- case PayloadType::UNCOMPRESSABLE: {
- std::ifstream rnd_file(kRandomFile);
- GPR_ASSERT(rnd_file.good());
- for (int n = 0; n < response_stream_sizes[k]; n++) {
- GPR_ASSERT(response.payload().body()[n] == (char)rnd_file.get());
- }
- } break;
- default:
- GPR_ASSERT(false);
- }
-
- // Compression related checks.
- GPR_ASSERT(request.response_compression() ==
- GetInteropCompressionTypeFromCompressionAlgorithm(
- inspector.GetCallCompressionAlgorithm()));
- if (request.response_compression() == NONE) {
- GPR_ASSERT(
- !(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
- } else if (request.response_type() == PayloadType::COMPRESSABLE) {
- // requested compression and compressable response => results should
- // always be compressed.
- GPR_ASSERT(inspector.GetMessageFlags() &
- GRPC_WRITE_INTERNAL_COMPRESS);
- }
-
- ++k;
- }
-
- GPR_ASSERT(response_stream_sizes.size() == k);
- Status s = stream->Finish();
-
- AssertOkOrPrintErrorStatus(s);
- gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix);
- gpr_free(log_suffix);
+bool InteropClient::DoClientCompressedStreaming() {
+ // Probing for compression-checks support.
+ ClientContext probe_context;
+ StreamingInputCallRequest probe_req;
+ StreamingInputCallResponse probe_res;
+
+ probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
+ probe_req.mutable_expect_compressed()->set_value(true); // lies!
+ probe_req.mutable_payload()->set_body(grpc::string(27182, '\0'));
+
+ gpr_log(GPR_DEBUG, "Sending probe for compressed streaming request.");
+
+ std::unique_ptr<ClientWriter<StreamingInputCallRequest>> probe_stream(
+ serviceStub_.Get()->StreamingInputCall(&probe_context, &probe_res));
+
+ if (!probe_stream->Write(probe_req)) {
+ gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
+ return TransientFailureOrAbort();
+ }
+ Status s = probe_stream->Finish();
+ if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
+ // The server isn't able to evaluate incoming compression, making the rest
+ // of this test moot.
+ gpr_log(GPR_DEBUG, "Compressed streaming request probe failed");
+ return false;
+ }
+ gpr_log(GPR_DEBUG,
+ "Compressed streaming request probe succeeded. Proceeding.");
+
+ ClientContext context;
+ StreamingInputCallRequest request;
+ StreamingInputCallResponse response;
+
+ context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
+ std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
+ serviceStub_.Get()->StreamingInputCall(&context, &response));
+
+ request.mutable_payload()->set_body(grpc::string(27182, '\0'));
+ request.mutable_expect_compressed()->set_value(true);
+ gpr_log(GPR_DEBUG, "Sending streaming request with compression enabled");
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
+ return TransientFailureOrAbort();
+ }
+
+ WriteOptions wopts;
+ wopts.set_no_compression();
+ request.mutable_payload()->set_body(grpc::string(45904, '\0'));
+ request.mutable_expect_compressed()->set_value(false);
+ gpr_log(GPR_DEBUG, "Sending streaming request with compression disabled");
+ if (!stream->Write(request, wopts)) {
+ gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
+ return TransientFailureOrAbort();
+ }
+ GPR_ASSERT(stream->WritesDone());
+
+ s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
+ return true;
+}
+
+bool InteropClient::DoServerCompressedStreaming() {
+ const std::vector<bool> compressions = {true, false};
+ const std::vector<int> sizes = {31415, 92653};
+
+ ClientContext context;
+ InteropClientContextInspector inspector(context);
+ StreamingOutputCallRequest request;
+
+ GPR_ASSERT(compressions.size() == sizes.size());
+ for (size_t i = 0; i < sizes.size(); i++) {
+ char* log_suffix;
+ gpr_asprintf(&log_suffix, "(compression=%s; size=%d)",
+ compressions[i] ? "true" : "false", sizes[i]);
+
+ gpr_log(GPR_DEBUG, "Sending request streaming rpc %s.", log_suffix);
+ gpr_free(log_suffix);
+
+ ResponseParameters* const response_parameter =
+ request.add_response_parameters();
+ response_parameter->mutable_compressed()->set_value(compressions[i]);
+ response_parameter->set_size(sizes[i]);
+ }
+ std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
+ serviceStub_.Get()->StreamingOutputCall(&context, request));
+
+ size_t k = 0;
+ StreamingOutputCallResponse response;
+ while (stream->Read(&response)) {
+ // Payload size checks.
+ GPR_ASSERT(response.payload().body() ==
+ grpc::string(request.response_parameters(k).size(), '\0'));
+
+ // Compression checks.
+ GPR_ASSERT(request.response_parameters(k).has_compressed());
+ if (request.response_parameters(k).compressed().value()) {
+ GPR_ASSERT(inspector.GetCallCompressionAlgorithm() > GRPC_COMPRESS_NONE);
+ GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
+ } else {
+ // requested *no* compression.
+ GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
}
+ ++k;
+ }
+
+ if (k < sizes.size()) {
+ // stream->Read() failed before reading all the expected messages. This
+ // is most likely due to a connection failure.
+ gpr_log(GPR_ERROR,
+ "%s(): Responses read (k=%" PRIuPTR
+ ") is less than the expected number of messages (%" PRIuPTR ").",
+ __func__, k, sizes.size());
+ return TransientFailureOrAbort();
+ }
+
+ Status s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
}
+ return true;
}
-void InteropClient::DoResponseStreamingWithSlowConsumer() {
- gpr_log(GPR_DEBUG, "Receiving response steaming rpc with slow consumer ...");
+bool InteropClient::DoResponseStreamingWithSlowConsumer() {
+ gpr_log(GPR_DEBUG, "Receiving response streaming rpc with slow consumer ...");
ClientContext context;
StreamingOutputCallRequest request;
@@ -464,14 +612,26 @@ void InteropClient::DoResponseStreamingWithSlowConsumer() {
usleep(kReceiveDelayMilliSeconds * 1000);
++i;
}
- GPR_ASSERT(kNumResponseMessages == i);
+
+ if (i < kNumResponseMessages) {
+ gpr_log(GPR_ERROR,
+ "DoResponseStreamingWithSlowConsumer(): Responses read (i=%d) is "
+ "less than the expected messages (i.e kNumResponseMessages = %d)",
+ i, kNumResponseMessages);
+
+ return TransientFailureOrAbort();
+ }
+
Status s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
- AssertOkOrPrintErrorStatus(s);
gpr_log(GPR_DEBUG, "Response streaming done.");
+ return true;
}
-void InteropClient::DoHalfDuplex() {
+bool InteropClient::DoHalfDuplex() {
gpr_log(GPR_DEBUG, "Sending half-duplex streaming rpc ...");
ClientContext context;
@@ -483,7 +643,11 @@ void InteropClient::DoHalfDuplex() {
ResponseParameters* response_parameter = request.add_response_parameters();
for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
response_parameter->set_size(response_stream_sizes[i]);
- GPR_ASSERT(stream->Write(request));
+
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "DoHalfDuplex(): stream->Write() failed. i=%d", i);
+ return TransientFailureOrAbort();
+ }
}
stream->WritesDone();
@@ -494,13 +658,27 @@ void InteropClient::DoHalfDuplex() {
grpc::string(response_stream_sizes[i], '\0'));
++i;
}
- GPR_ASSERT(response_stream_sizes.size() == i);
+
+ if (i < response_stream_sizes.size()) {
+ // stream->Read() failed before reading all the expected messages. This is
+ // most likely due to a connection failure
+ gpr_log(GPR_ERROR,
+ "DoHalfDuplex(): Responses read (i=%d) are less than the expected "
+ "number of messages response_stream_sizes.size() (%" PRIuPTR ")",
+ i, response_stream_sizes.size());
+ return TransientFailureOrAbort();
+ }
+
Status s = stream->Finish();
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "Half-duplex streaming rpc done.");
+ return true;
}
-void InteropClient::DoPingPong() {
+bool InteropClient::DoPingPong() {
gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
ClientContext context;
@@ -509,28 +687,43 @@ void InteropClient::DoPingPong() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
Payload* payload = request.mutable_payload();
StreamingOutputCallResponse response;
+
for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
response_parameter->set_size(response_stream_sizes[i]);
payload->set_body(grpc::string(request_stream_sizes[i], '\0'));
- GPR_ASSERT(stream->Write(request));
- GPR_ASSERT(stream->Read(&response));
+
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "DoPingPong(): stream->Write() failed. i: %d", i);
+ return TransientFailureOrAbort();
+ }
+
+ if (!stream->Read(&response)) {
+ gpr_log(GPR_ERROR, "DoPingPong(): stream->Read() failed. i:%d", i);
+ return TransientFailureOrAbort();
+ }
+
GPR_ASSERT(response.payload().body() ==
grpc::string(response_stream_sizes[i], '\0'));
}
stream->WritesDone();
+
GPR_ASSERT(!stream->Read(&response));
+
Status s = stream->Finish();
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "Ping pong streaming done.");
+ return true;
}
-void InteropClient::DoCancelAfterBegin() {
- gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
+bool InteropClient::DoCancelAfterBegin() {
+ gpr_log(GPR_DEBUG, "Sending request streaming rpc ...");
ClientContext context;
StreamingInputCallRequest request;
@@ -542,11 +735,16 @@ void InteropClient::DoCancelAfterBegin() {
gpr_log(GPR_DEBUG, "Trying to cancel...");
context.TryCancel();
Status s = stream->Finish();
- GPR_ASSERT(s.error_code() == StatusCode::CANCELLED);
+
+ if (!AssertStatusCode(s, StatusCode::CANCELLED)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "Canceling streaming done.");
+ return true;
}
-void InteropClient::DoCancelAfterFirstResponse() {
+bool InteropClient::DoCancelAfterFirstResponse() {
gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
ClientContext context;
@@ -555,22 +753,31 @@ void InteropClient::DoCancelAfterFirstResponse() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
response_parameter->set_size(31415);
request.mutable_payload()->set_body(grpc::string(27182, '\0'));
StreamingOutputCallResponse response;
- GPR_ASSERT(stream->Write(request));
- GPR_ASSERT(stream->Read(&response));
+
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Write() failed");
+ return TransientFailureOrAbort();
+ }
+
+ if (!stream->Read(&response)) {
+ gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Read failed");
+ return TransientFailureOrAbort();
+ }
GPR_ASSERT(response.payload().body() == grpc::string(31415, '\0'));
+
gpr_log(GPR_DEBUG, "Trying to cancel...");
context.TryCancel();
Status s = stream->Finish();
gpr_log(GPR_DEBUG, "Canceling pingpong streaming done.");
+ return true;
}
-void InteropClient::DoTimeoutOnSleepingServer() {
+bool InteropClient::DoTimeoutOnSleepingServer() {
gpr_log(GPR_DEBUG,
"Sending Ping Pong streaming rpc with a short deadline...");
@@ -587,11 +794,15 @@ void InteropClient::DoTimeoutOnSleepingServer() {
stream->Write(request);
Status s = stream->Finish();
- GPR_ASSERT(s.error_code() == StatusCode::DEADLINE_EXCEEDED);
+ if (!AssertStatusCode(s, StatusCode::DEADLINE_EXCEEDED)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "Pingpong streaming timeout done.");
+ return true;
}
-void InteropClient::DoEmptyStream() {
+bool InteropClient::DoEmptyStream() {
gpr_log(GPR_DEBUG, "Starting empty_stream.");
ClientContext context;
@@ -601,12 +812,17 @@ void InteropClient::DoEmptyStream() {
stream->WritesDone();
StreamingOutputCallResponse response;
GPR_ASSERT(stream->Read(&response) == false);
+
Status s = stream->Finish();
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
gpr_log(GPR_DEBUG, "empty_stream done.");
+ return true;
}
-void InteropClient::DoStatusWithMessage() {
+bool InteropClient::DoStatusWithMessage() {
gpr_log(GPR_DEBUG,
"Sending RPC with a request for status code 2 and message");
@@ -620,12 +836,16 @@ void InteropClient::DoStatusWithMessage() {
Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
- GPR_ASSERT(s.error_code() == grpc::StatusCode::UNKNOWN);
+ if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN)) {
+ return false;
+ }
+
GPR_ASSERT(s.error_message() == test_msg);
gpr_log(GPR_DEBUG, "Done testing Status and Message");
+ return true;
}
-void InteropClient::DoCustomMetadata() {
+bool InteropClient::DoCustomMetadata() {
const grpc::string kEchoInitialMetadataKey("x-grpc-test-echo-initial");
const grpc::string kInitialMetadataValue("test_initial_metadata_value");
const grpc::string kEchoTrailingBinMetadataKey(
@@ -645,7 +865,10 @@ void InteropClient::DoCustomMetadata() {
request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
const auto& server_initial_metadata = context.GetServerInitialMetadata();
auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
GPR_ASSERT(iter != server_initial_metadata.end());
@@ -669,20 +892,34 @@ void InteropClient::DoCustomMetadata() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
response_parameter->set_size(kLargeResponseSize);
grpc::string payload(kLargeRequestSize, '\0');
request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
StreamingOutputCallResponse response;
- GPR_ASSERT(stream->Write(request));
+
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Write() failed");
+ return TransientFailureOrAbort();
+ }
+
stream->WritesDone();
- GPR_ASSERT(stream->Read(&response));
+
+ if (!stream->Read(&response)) {
+ gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Read() failed");
+ return TransientFailureOrAbort();
+ }
+
GPR_ASSERT(response.payload().body() ==
grpc::string(kLargeResponseSize, '\0'));
+
GPR_ASSERT(!stream->Read(&response));
+
Status s = stream->Finish();
- AssertOkOrPrintErrorStatus(s);
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
const auto& server_initial_metadata = context.GetServerInitialMetadata();
auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
GPR_ASSERT(iter != server_initial_metadata.end());
@@ -695,6 +932,8 @@ void InteropClient::DoCustomMetadata() {
gpr_log(GPR_DEBUG, "Done testing stream with custom metadata");
}
+
+ return true;
}
} // namespace testing
diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h
index a3794fd93f..eb886fcb7e 100644
--- a/test/cpp/interop/interop_client.h
+++ b/test/cpp/interop/interop_client.h
@@ -45,47 +45,50 @@ namespace grpc {
namespace testing {
// Function pointer for custom checks.
-using CheckerFn =
- std::function<void(const InteropClientContextInspector&,
- const SimpleRequest*, const SimpleResponse*)>;
+typedef std::function<void(const InteropClientContextInspector&,
+ const SimpleRequest*, const SimpleResponse*)>
+ CheckerFn;
class InteropClient {
public:
- explicit InteropClient(std::shared_ptr<Channel> channel);
- explicit InteropClient(
- std::shared_ptr<Channel> channel,
- bool new_stub_every_test_case); // If new_stub_every_test_case is true,
- // a new TestService::Stub object is
- // created for every test case below
+ /// If new_stub_every_test_case is true, a new TestService::Stub object is
+ /// created for every test case
+ /// If do_not_abort_on_transient_failures is true, abort() is not called in
+ /// case of transient failures (like connection failures)
+ explicit InteropClient(std::shared_ptr<Channel> channel,
+ bool new_stub_every_test_case,
+ bool do_not_abort_on_transient_failures);
~InteropClient() {}
void Reset(std::shared_ptr<Channel> channel);
- void DoEmpty();
- void DoLargeUnary();
- void DoLargeCompressedUnary();
- void DoPingPong();
- void DoHalfDuplex();
- void DoRequestStreaming();
- void DoResponseStreaming();
- void DoResponseCompressedStreaming();
- void DoResponseStreamingWithSlowConsumer();
- void DoCancelAfterBegin();
- void DoCancelAfterFirstResponse();
- void DoTimeoutOnSleepingServer();
- void DoEmptyStream();
- void DoStatusWithMessage();
- void DoCustomMetadata();
+ bool DoEmpty();
+ bool DoLargeUnary();
+ bool DoServerCompressedUnary();
+ bool DoClientCompressedUnary();
+ bool DoPingPong();
+ bool DoHalfDuplex();
+ bool DoRequestStreaming();
+ bool DoResponseStreaming();
+ bool DoServerCompressedStreaming();
+ bool DoClientCompressedStreaming();
+ bool DoResponseStreamingWithSlowConsumer();
+ bool DoCancelAfterBegin();
+ bool DoCancelAfterFirstResponse();
+ bool DoTimeoutOnSleepingServer();
+ bool DoEmptyStream();
+ bool DoStatusWithMessage();
+ bool DoCustomMetadata();
// Auth tests.
// username is a string containing the user email
- void DoJwtTokenCreds(const grpc::string& username);
- void DoComputeEngineCreds(const grpc::string& default_service_account,
+ bool DoJwtTokenCreds(const grpc::string& username);
+ bool DoComputeEngineCreds(const grpc::string& default_service_account,
const grpc::string& oauth_scope);
// username the GCE default service account email
- void DoOauth2AuthToken(const grpc::string& username,
+ bool DoOauth2AuthToken(const grpc::string& username,
const grpc::string& oauth_scope);
// username is a string containing the user email
- void DoPerRpcCreds(const grpc::string& json_key);
+ bool DoPerRpcCreds(const grpc::string& json_key);
private:
class ServiceStub {
@@ -105,13 +108,18 @@ class InteropClient {
// Get() call
};
- void PerformLargeUnary(SimpleRequest* request, SimpleResponse* response);
+ bool PerformLargeUnary(SimpleRequest* request, SimpleResponse* response);
/// Run \a custom_check_fn as an additional check.
- void PerformLargeUnary(SimpleRequest* request, SimpleResponse* response,
+ bool PerformLargeUnary(SimpleRequest* request, SimpleResponse* response,
CheckerFn custom_checks_fn);
- void AssertOkOrPrintErrorStatus(const Status& s);
+ bool AssertStatusOk(const Status& s);
+ bool AssertStatusCode(const Status& s, StatusCode expected_code);
+ bool TransientFailureOrAbort();
ServiceStub serviceStub_;
+
+ /// If true, abort() is not called for transient failures
+ bool do_not_abort_on_transient_failures_;
};
} // namespace testing
diff --git a/test/cpp/interop/server_main.cc b/test/cpp/interop/interop_server.cc
index 889874fe49..ebef0002a3 100644
--- a/test/cpp/interop/server_main.cc
+++ b/test/cpp/interop/interop_server.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -48,6 +48,7 @@
#include <grpc/support/log.h>
#include <grpc/support/useful.h>
+#include "src/core/lib/transport/byte_stream.h"
#include "src/proto/grpc/testing/empty.grpc.pb.h"
#include "src/proto/grpc/testing/messages.grpc.pb.h"
#include "src/proto/grpc/testing/test.grpc.pb.h"
@@ -64,10 +65,10 @@ using grpc::ServerCredentials;
using grpc::ServerReader;
using grpc::ServerReaderWriter;
using grpc::ServerWriter;
+using grpc::WriteOptions;
using grpc::SslServerCredentialsOptions;
using grpc::testing::InteropServerContextInspector;
using grpc::testing::Payload;
-using grpc::testing::PayloadType;
using grpc::testing::SimpleRequest;
using grpc::testing::SimpleResponse;
using grpc::testing::StreamingInputCallRequest;
@@ -78,7 +79,6 @@ using grpc::testing::TestService;
using grpc::Status;
static bool got_sigint = false;
-static const char* kRandomFile = "test/cpp/interop/rnd.dat";
const char kEchoInitialMetadataKey[] = "x-grpc-test-echo-initial";
const char kEchoTrailingBinMetadataKey[] = "x-grpc-test-echo-trailing-bin";
@@ -110,50 +110,41 @@ void MaybeEchoMetadata(ServerContext* context) {
}
}
-bool SetPayload(PayloadType type, int size, Payload* payload) {
- PayloadType response_type;
- if (type == PayloadType::RANDOM) {
- response_type =
- rand() & 0x1 ? PayloadType::COMPRESSABLE : PayloadType::UNCOMPRESSABLE;
- } else {
- response_type = type;
- }
- payload->set_type(response_type);
- switch (response_type) {
- case PayloadType::COMPRESSABLE: {
- std::unique_ptr<char[]> body(new char[size]());
- payload->set_body(body.get(), size);
- } break;
- case PayloadType::UNCOMPRESSABLE: {
- std::unique_ptr<char[]> body(new char[size]());
- std::ifstream rnd_file(kRandomFile);
- GPR_ASSERT(rnd_file.good());
- rnd_file.read(body.get(), size);
- GPR_ASSERT(!rnd_file.eof()); // Requested more rnd bytes than available
- payload->set_body(body.get(), size);
- } break;
- default:
- GPR_ASSERT(false);
- }
+bool SetPayload(int size, Payload* payload) {
+ std::unique_ptr<char[]> body(new char[size]());
+ payload->set_body(body.get(), size);
return true;
}
-template <typename RequestType>
-void SetResponseCompression(ServerContext* context,
- const RequestType& request) {
- switch (request.response_compression()) {
- case grpc::testing::NONE:
- context->set_compression_algorithm(GRPC_COMPRESS_NONE);
- break;
- case grpc::testing::GZIP:
- context->set_compression_algorithm(GRPC_COMPRESS_GZIP);
- break;
- case grpc::testing::DEFLATE:
- context->set_compression_algorithm(GRPC_COMPRESS_DEFLATE);
- break;
- default:
- abort();
+bool CheckExpectedCompression(const ServerContext& context,
+ const bool compression_expected) {
+ const InteropServerContextInspector inspector(context);
+ const grpc_compression_algorithm received_compression =
+ inspector.GetCallCompressionAlgorithm();
+
+ if (compression_expected) {
+ if (received_compression == GRPC_COMPRESS_NONE) {
+ // Expected some compression, got NONE. This is an error.
+ gpr_log(GPR_ERROR,
+ "Expected compression but got uncompressed request from client.");
+ return false;
+ }
+ if (!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS)) {
+ gpr_log(GPR_ERROR,
+ "Failure: Requested compression in a compressable request, but "
+ "compression bit in message flags not set.");
+ return false;
+ }
+ } else {
+ // Didn't expect compression -> make sure the request is uncompressed
+ if (inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS) {
+ gpr_log(GPR_ERROR,
+ "Failure: Didn't requested compression, but compression bit in "
+ "message flags set.");
+ return false;
+ }
}
+ return true;
}
class TestServiceImpl : public TestService::Service {
@@ -167,11 +158,26 @@ class TestServiceImpl : public TestService::Service {
Status UnaryCall(ServerContext* context, const SimpleRequest* request,
SimpleResponse* response) {
MaybeEchoMetadata(context);
- SetResponseCompression(context, *request);
+ if (request->has_response_compressed()) {
+ const bool compression_requested = request->response_compressed().value();
+ gpr_log(GPR_DEBUG, "Request for compression (%s) present for %s",
+ compression_requested ? "enabled" : "disabled", __func__);
+ if (compression_requested) {
+ // Any level would do, let's go for HIGH because we are overachievers.
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
+ } else {
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_NONE);
+ }
+ }
+ if (!CheckExpectedCompression(*context,
+ request->expect_compressed().value())) {
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Compressed request expectation not met.");
+ }
if (request->response_size() > 0) {
- if (!SetPayload(request->response_type(), request->response_size(),
- response->mutable_payload())) {
- return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+ if (!SetPayload(request->response_size(), response->mutable_payload())) {
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Error creating payload.");
}
}
@@ -187,17 +193,36 @@ class TestServiceImpl : public TestService::Service {
Status StreamingOutputCall(
ServerContext* context, const StreamingOutputCallRequest* request,
ServerWriter<StreamingOutputCallResponse>* writer) {
- SetResponseCompression(context, *request);
StreamingOutputCallResponse response;
bool write_success = true;
for (int i = 0; write_success && i < request->response_parameters_size();
i++) {
- if (!SetPayload(request->response_type(),
- request->response_parameters(i).size(),
+ if (!SetPayload(request->response_parameters(i).size(),
response.mutable_payload())) {
- return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Error creating payload.");
}
- write_success = writer->Write(response);
+ WriteOptions wopts;
+ if (request->response_parameters(i).has_compressed()) {
+ // Compress by default. Disabled on a per-message basis.
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
+ const bool compression_requested =
+ request->response_parameters(i).compressed().value();
+ gpr_log(GPR_DEBUG, "Request for compression (%s) present for %s",
+ compression_requested ? "enabled" : "disabled", __func__);
+ if (!compression_requested) {
+ wopts.set_no_compression();
+ } // else, compression is already enabled via the context.
+ }
+ int time_us;
+ if ((time_us = request->response_parameters(i).interval_us()) > 0) {
+ // Sleep before response if needed
+ gpr_timespec sleep_time =
+ gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+ gpr_time_from_micros(time_us, GPR_TIMESPAN));
+ gpr_sleep_until(sleep_time);
+ }
+ write_success = writer->Write(response, wopts);
}
if (write_success) {
return Status::OK;
@@ -212,6 +237,11 @@ class TestServiceImpl : public TestService::Service {
StreamingInputCallRequest request;
int aggregated_payload_size = 0;
while (reader->Read(&request)) {
+ if (!CheckExpectedCompression(*context,
+ request.expect_compressed().value())) {
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Compressed request expectation not met.");
+ }
if (request.has_payload()) {
aggregated_payload_size += request.payload().body().size();
}
@@ -229,11 +259,18 @@ class TestServiceImpl : public TestService::Service {
StreamingOutputCallResponse response;
bool write_success = true;
while (write_success && stream->Read(&request)) {
- SetResponseCompression(context, request);
if (request.response_parameters_size() != 0) {
response.mutable_payload()->set_type(request.payload().type());
response.mutable_payload()->set_body(
grpc::string(request.response_parameters(0).size(), '\0'));
+ int time_us;
+ if ((time_us = request.response_parameters(0).interval_us()) > 0) {
+ // Sleep before response if needed
+ gpr_timespec sleep_time =
+ gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+ gpr_time_from_micros(time_us, GPR_TIMESPAN));
+ gpr_sleep_until(sleep_time);
+ }
write_success = stream->Write(response);
}
}
diff --git a/test/cpp/interop/metrics_client.cc b/test/cpp/interop/metrics_client.cc
index cc304f2e89..7a0cb994df 100644
--- a/test/cpp/interop/metrics_client.cc
+++ b/test/cpp/interop/metrics_client.cc
@@ -42,13 +42,15 @@
#include "test/cpp/util/metrics_server.h"
#include "test/cpp/util/test_config.h"
-DEFINE_string(metrics_server_address, "",
+int kDeadlineSecs = 10;
+
+DEFINE_string(metrics_server_address, "localhost:8081",
"The metrics server addresses in the fomrat <hostname>:<port>");
+DEFINE_int32(deadline_secs, kDeadlineSecs,
+ "The deadline (in seconds) for RCP call");
DEFINE_bool(total_only, false,
"If true, this prints only the total value of all gauges");
-int kDeadlineSecs = 10;
-
using grpc::testing::EmptyMessage;
using grpc::testing::GaugeResponse;
using grpc::testing::MetricsService;
@@ -56,12 +58,13 @@ using grpc::testing::MetricsServiceImpl;
// Prints the values of all Gauges (unless total_only is set to 'true' in which
// case this only prints the sum of all gauge values).
-bool PrintMetrics(std::unique_ptr<MetricsService::Stub> stub, bool total_only) {
+bool PrintMetrics(std::unique_ptr<MetricsService::Stub> stub, bool total_only,
+ int deadline_secs) {
grpc::ClientContext context;
EmptyMessage message;
std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(kDeadlineSecs);
+ std::chrono::system_clock::now() + std::chrono::seconds(deadline_secs);
context.set_deadline(deadline);
@@ -73,7 +76,7 @@ bool PrintMetrics(std::unique_ptr<MetricsService::Stub> stub, bool total_only) {
while (reader->Read(&gauge_response)) {
if (gauge_response.value_case() == GaugeResponse::kLongValue) {
if (!total_only) {
- gpr_log(GPR_INFO, "%s: %ld", gauge_response.name().c_str(),
+ gpr_log(GPR_INFO, "%s: %lld", gauge_response.name().c_str(),
gauge_response.long_value());
}
overall_qps += gauge_response.long_value();
@@ -108,7 +111,8 @@ int main(int argc, char** argv) {
std::shared_ptr<grpc::Channel> channel(grpc::CreateChannel(
FLAGS_metrics_server_address, grpc::InsecureChannelCredentials()));
- if (!PrintMetrics(MetricsService::NewStub(channel), FLAGS_total_only)) {
+ if (!PrintMetrics(MetricsService::NewStub(channel), FLAGS_total_only,
+ FLAGS_deadline_secs)) {
return 1;
}
diff --git a/test/cpp/interop/rnd.dat b/test/cpp/interop/rnd.dat
deleted file mode 100644
index 8c7f38f9e0..0000000000
--- a/test/cpp/interop/rnd.dat
+++ /dev/null
Binary files differ
diff --git a/test/cpp/interop/server_helper.cc b/test/cpp/interop/server_helper.cc
index c6d891ad71..8b0b511bcb 100644
--- a/test/cpp/interop/server_helper.cc
+++ b/test/cpp/interop/server_helper.cc
@@ -72,6 +72,10 @@ uint32_t InteropServerContextInspector::GetEncodingsAcceptedByClient() const {
return grpc_call_test_only_get_encodings_accepted_by_peer(context_.call_);
}
+uint32_t InteropServerContextInspector::GetMessageFlags() const {
+ return grpc_call_test_only_get_message_flags(context_.call_);
+}
+
std::shared_ptr<const AuthContext>
InteropServerContextInspector::GetAuthContext() const {
return context_.auth_context();
diff --git a/test/cpp/interop/server_helper.h b/test/cpp/interop/server_helper.h
index 12865e4032..a1da14a4c8 100644
--- a/test/cpp/interop/server_helper.h
+++ b/test/cpp/interop/server_helper.h
@@ -54,6 +54,7 @@ class InteropServerContextInspector {
bool IsCancelled() const;
grpc_compression_algorithm GetCallCompressionAlgorithm() const;
uint32_t GetEncodingsAcceptedByClient() const;
+ uint32_t GetMessageFlags() const;
private:
const ::grpc::ServerContext& context_;
diff --git a/test/cpp/interop/stress_interop_client.cc b/test/cpp/interop/stress_interop_client.cc
index f287a5aa3b..1d5fc80cf2 100644
--- a/test/cpp/interop/stress_interop_client.cc
+++ b/test/cpp/interop/stress_interop_client.cc
@@ -84,11 +84,12 @@ 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 sleep_duration_ms, bool do_not_abort_on_transient_failures)
: test_id_(test_id),
server_address_(server_address),
channel_(channel),
- interop_client_(new InteropClient(channel, false)),
+ interop_client_(new InteropClient(channel, false,
+ do_not_abort_on_transient_failures)),
test_selector_(test_selector),
test_duration_secs_(test_duration_secs),
sleep_duration_ms_(sleep_duration_ms) {}
@@ -126,31 +127,75 @@ void StressTestInteropClient::MainLoop(std::shared_ptr<QpsGauge> qps_gauge) {
}
}
-// TODO(sree): Add all interop tests
-void StressTestInteropClient::RunTest(TestCaseType test_case) {
+bool StressTestInteropClient::RunTest(TestCaseType test_case) {
+ bool is_success = false;
switch (test_case) {
case EMPTY_UNARY: {
- interop_client_->DoEmpty();
+ is_success = interop_client_->DoEmpty();
break;
}
case LARGE_UNARY: {
- interop_client_->DoLargeUnary();
+ is_success = interop_client_->DoLargeUnary();
break;
}
- case LARGE_COMPRESSED_UNARY: {
- interop_client_->DoLargeCompressedUnary();
+ case CLIENT_COMPRESSED_UNARY: {
+ is_success = interop_client_->DoClientCompressedUnary();
+ break;
+ }
+ case CLIENT_COMPRESSED_STREAMING: {
+ is_success = interop_client_->DoClientCompressedStreaming();
break;
}
case CLIENT_STREAMING: {
- interop_client_->DoRequestStreaming();
+ is_success = interop_client_->DoRequestStreaming();
break;
}
case SERVER_STREAMING: {
- interop_client_->DoResponseStreaming();
+ is_success = interop_client_->DoResponseStreaming();
+ break;
+ }
+ case SERVER_COMPRESSED_UNARY: {
+ is_success = interop_client_->DoServerCompressedUnary();
+ break;
+ }
+ case SERVER_COMPRESSED_STREAMING: {
+ is_success = interop_client_->DoServerCompressedStreaming();
+ break;
+ }
+ case SLOW_CONSUMER: {
+ is_success = interop_client_->DoResponseStreamingWithSlowConsumer();
+ break;
+ }
+ case HALF_DUPLEX: {
+ is_success = interop_client_->DoHalfDuplex();
+ break;
+ }
+ case PING_PONG: {
+ is_success = interop_client_->DoPingPong();
+ break;
+ }
+ case CANCEL_AFTER_BEGIN: {
+ is_success = interop_client_->DoCancelAfterBegin();
+ break;
+ }
+ case CANCEL_AFTER_FIRST_RESPONSE: {
+ is_success = interop_client_->DoCancelAfterFirstResponse();
+ break;
+ }
+ case TIMEOUT_ON_SLEEPING_SERVER: {
+ is_success = interop_client_->DoTimeoutOnSleepingServer();
break;
}
case EMPTY_STREAM: {
- interop_client_->DoEmptyStream();
+ is_success = interop_client_->DoEmptyStream();
+ break;
+ }
+ case STATUS_CODE_AND_MESSAGE: {
+ is_success = interop_client_->DoStatusWithMessage();
+ break;
+ }
+ case CUSTOM_METADATA: {
+ is_success = interop_client_->DoCustomMetadata();
break;
}
default: {
@@ -159,6 +204,8 @@ void StressTestInteropClient::RunTest(TestCaseType test_case) {
break;
}
}
+
+ return is_success;
}
} // namespace testing
diff --git a/test/cpp/interop/stress_interop_client.h b/test/cpp/interop/stress_interop_client.h
index cb0cd98821..cf6a713473 100644
--- a/test/cpp/interop/stress_interop_client.h
+++ b/test/cpp/interop/stress_interop_client.h
@@ -49,24 +49,45 @@ namespace testing {
using std::pair;
using std::vector;
-// TODO(sreek): Add more test cases here in future
enum TestCaseType {
UNKNOWN_TEST = -1,
- EMPTY_UNARY = 0,
- LARGE_UNARY = 1,
- LARGE_COMPRESSED_UNARY = 2,
- CLIENT_STREAMING = 3,
- SERVER_STREAMING = 4,
- EMPTY_STREAM = 5
+ EMPTY_UNARY,
+ LARGE_UNARY,
+ CLIENT_COMPRESSED_UNARY,
+ CLIENT_COMPRESSED_STREAMING,
+ CLIENT_STREAMING,
+ SERVER_STREAMING,
+ SERVER_COMPRESSED_UNARY,
+ SERVER_COMPRESSED_STREAMING,
+ SLOW_CONSUMER,
+ HALF_DUPLEX,
+ PING_PONG,
+ CANCEL_AFTER_BEGIN,
+ CANCEL_AFTER_FIRST_RESPONSE,
+ TIMEOUT_ON_SLEEPING_SERVER,
+ EMPTY_STREAM,
+ STATUS_CODE_AND_MESSAGE,
+ CUSTOM_METADATA
};
const vector<pair<TestCaseType, grpc::string>> kTestCaseList = {
{EMPTY_UNARY, "empty_unary"},
{LARGE_UNARY, "large_unary"},
- {LARGE_COMPRESSED_UNARY, "large_compressed_unary"},
+ {CLIENT_COMPRESSED_UNARY, "client_compressed_unary"},
+ {CLIENT_COMPRESSED_STREAMING, "client_compressed_streaming"},
{CLIENT_STREAMING, "client_streaming"},
{SERVER_STREAMING, "server_streaming"},
- {EMPTY_STREAM, "empty_stream"}};
+ {SERVER_COMPRESSED_UNARY, "server_compressed_unary"},
+ {SERVER_COMPRESSED_STREAMING, "server_compressed_streaming"},
+ {SLOW_CONSUMER, "slow_consumer"},
+ {HALF_DUPLEX, "half_duplex"},
+ {PING_PONG, "ping_pong"},
+ {CANCEL_AFTER_BEGIN, "cancel_after_begin"},
+ {CANCEL_AFTER_FIRST_RESPONSE, "cancel_after_first_response"},
+ {TIMEOUT_ON_SLEEPING_SERVER, "timeout_on_sleeping_server"},
+ {EMPTY_STREAM, "empty_stream"},
+ {STATUS_CODE_AND_MESSAGE, "status_code_and_message"},
+ {CUSTOM_METADATA, "custom_metadata"}};
class WeightedRandomTestSelector {
public:
@@ -87,14 +108,15 @@ 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 test_duration_secs, long sleep_duration_ms,
+ bool do_not_abort_on_transient_failures);
// The main function. Use this as the thread entry point.
// 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);
+ bool RunTest(TestCaseType test_case);
int test_id_;
const grpc::string& server_address_;
diff --git a/test/cpp/interop/stress_test.cc b/test/cpp/interop/stress_test.cc
index d9e3fd25c5..7787931900 100644
--- a/test/cpp/interop/stress_test.cc
+++ b/test/cpp/interop/stress_test.cc
@@ -89,7 +89,16 @@ DEFINE_string(test_cases, "",
" large_compressed_unary\n"
" client_streaming\n"
" server_streaming\n"
+ " server_compressed_streaming\n"
+ " slow_consumer\n"
+ " half_duplex\n"
+ " ping_pong\n"
+ " cancel_after_begin\n"
+ " cancel_after_first_response\n"
+ " timeout_on_sleeping_server\n"
" empty_stream\n"
+ " status_code_and_message\n"
+ " custom_metadata\n"
" Example: \"empty_unary:20,large_unary:10,empty_stream:70\"\n"
" The above will execute 'empty_unary', 20% of the time,"
" 'large_unary', 10% of the time and 'empty_stream' the remaining"
@@ -101,6 +110,10 @@ DEFINE_int32(log_level, GPR_LOG_SEVERITY_INFO,
"The choices are: 0 (GPR_LOG_SEVERITY_DEBUG), 1 "
"(GPR_LOG_SEVERITY_INFO) and 2 (GPR_LOG_SEVERITY_ERROR)");
+DEFINE_bool(do_not_abort_on_transient_failures, true,
+ "If set to 'true', abort() is not called in case of transient "
+ "failures like temporary connection failures.");
+
using grpc::testing::kTestCaseList;
using grpc::testing::MetricsService;
using grpc::testing::MetricsServiceImpl;
@@ -189,6 +202,12 @@ void LogParameterInfo(const std::vector<grpc::string>& addresses,
gpr_log(GPR_INFO, "test_cases : %s", FLAGS_test_cases.c_str());
gpr_log(GPR_INFO, "sleep_duration_ms: %d", FLAGS_sleep_duration_ms);
gpr_log(GPR_INFO, "test_duration_secs: %d", FLAGS_test_duration_secs);
+ gpr_log(GPR_INFO, "num_channels_per_server: %d",
+ FLAGS_num_channels_per_server);
+ gpr_log(GPR_INFO, "num_stubs_per_channel: %d", FLAGS_num_stubs_per_channel);
+ gpr_log(GPR_INFO, "log_level: %d", FLAGS_log_level);
+ gpr_log(GPR_INFO, "do_not_abort_on_transient_failures: %s",
+ FLAGS_do_not_abort_on_transient_failures ? "true" : "false");
int num = 0;
for (auto it = addresses.begin(); it != addresses.end(); it++) {
@@ -272,7 +291,7 @@ 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_sleep_duration_ms, FLAGS_do_not_abort_on_transient_failures);
bool is_already_created = false;
// QpsGauge name
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-01 13:04:40 +0000