C++ MCS connection scaling interop test client side changes

Author: kannanjgithub

test/cpp/interop/client.cc

@@ -228,10 +228,14 @@ int main(int argc, char** argv) {
           factories.emplace_back(
               new grpc::testing::MetadataAndStatusLoggerInterceptorFactory());
         }
-        std::string service_config_json =
-            absl::GetFlag(FLAGS_service_config_json);
-        if (!service_config_json.empty()) {
-          arguments.SetServiceConfigJSON(service_config_json);
+        if (test_case == "mcs_cs") {
+          arguments.SetServiceConfigJSON("{\"connection_scaling\":{\"max_connections_per_subchannel\": 2}}");
+        } else {
+          std::string service_config_json =
+              absl::GetFlag(FLAGS_service_config_json);
+          if (!service_config_json.empty()) {
+            arguments.SetServiceConfigJSON(service_config_json);
+          }
         }
         return CreateChannelForTestCase(test_case, std::move(factories),
                                         arguments);
@@ -279,6 +283,9 @@ int main(int argc, char** argv) {
       std::bind(&grpc::testing::InteropClient::DoOrcaPerRpc, &client);
   actions["orca_oob"] =
       std::bind(&grpc::testing::InteropClient::DoOrcaOob, &client);
+  actions["mcs_cs"] =
+      std::bind(&grpc::testing::InteropClient::DoMcsConnectionScaling, &client);
+
   if (absl::GetFlag(FLAGS_use_tls)) {
     actions["compute_engine_creds"] =
         std::bind(&grpc::testing::InteropClient::DoComputeEngineCreds, &client,

test/cpp/interop/interop_client.cc

@@ -1361,6 +1361,86 @@ bool InteropClient::DoLongLivedChannelTest(int32_t soak_iterations,
   }
 }
 
+bool InteropClient::DoMcsConnectionScaling() {
+  VLOG(2) << "Sending Mcs connection scaling streaming rpc1 ...";
+
+  ClientContext context1;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      stream1(serviceStub_.Get()->FullDuplexCall(&context1));
+
+  StreamingOutputCallRequest request;
+  Payload* payload = request.mutable_payload();
+  payload->set_body("max concurrent streaming connection scaling");
+  StreamingOutputCallResponse response1;
+
+  if (!stream1->Write(request)) {
+    LOG(ERROR) << "DoMcsConnectionScaling(): stream1->Write() failed.";
+    return TransientFailureOrAbort();
+  }
+  std::string clientSocketAddressInCall1 = response1.payload().body();
+
+  VLOG(2) << "Sending Mcs connection scaling streaming rpc2 ...";
+
+  ClientContext context2;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      stream2(serviceStub_.Get()->FullDuplexCall(&context2));
+
+  StreamingOutputCallResponse response2;
+
+  if (!stream2->Write(request)) {
+    LOG(ERROR) << "DoMcsConnectionScaling(): stream2->Write() failed.";
+    return TransientFailureOrAbort();
+  }
+  std::string clientSocketAddressInCall2 = response2.payload().body();
+
+  // The same connection should have been used for both streams.
+  GRPC_CHECK(response1.payload().body() == response2.payload().body());
+
+  VLOG(2) << "Sending Mcs connection scaling streaming rpc3 ...";
+
+  ClientContext context3;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      stream3(serviceStub_.Get()->FullDuplexCall(&context3));
+
+  StreamingOutputCallResponse response3;
+
+  if (!stream3->Write(request)) {
+    LOG(ERROR) << "DoMcsConnectionScaling(): stream3->Write() failed.";
+    return TransientFailureOrAbort();
+  }
+  std::string clientSocketAddressInCall3 = response3.payload().body();
+
+  // A new connection should have been used for the 3rd stream.
+  GRPC_CHECK(response1.payload().body() != response3.payload().body());
+
+  stream1->WritesDone();
+  GRPC_CHECK(!stream1->Read(&response1));
+  Status s = stream1->Finish();
+  if (!AssertStatusOk(s, context1.debug_error_string())) {
+    return false;
+  }
+
+  stream2->WritesDone();
+  GRPC_CHECK(!stream2->Read(&response2));
+  s = stream2->Finish();
+  if (!AssertStatusOk(s, context2.debug_error_string())) {
+    return false;
+  }
+
+  stream3->WritesDone();
+  GRPC_CHECK(!stream3->Read(&response3));
+  s = stream3->Finish();
+  if (!AssertStatusOk(s, context3.debug_error_string())) {
+    return false;
+  }
+
+  VLOG(2) << "Mcs connection scaling done.";
+  return true;
+}
+
 bool InteropClient::DoUnimplementedService() {
   VLOG(2) << "Sending a request for an unimplemented service...";
 

test/cpp/interop/interop_client.h

@@ -78,6 +78,7 @@ class InteropClient {
   bool DoPickFirstUnary();
   bool DoOrcaPerRpc();
   bool DoOrcaOob();
+  bool DoMcsConnectionScaling();
 
   // The following interop test are not yet part of the interop spec, and are
   // not implemented cross-language. They are considered experimental for now,