Skip to content

Commit c88285f

Browse files
kannanjgithubkannanjgithub
authored
MCS connection scaling interop tests for Java (grpc#12651)
Max Concurrent Streams connection scaling interop test client and server for Java.
1 parent 358f086 commit c88285f

6 files changed

Lines changed: 161 additions & 31 deletions

File tree

‎interop-testing/src/main/java/io/grpc/testing/integration/TestCases.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,8 @@ public enum TestCases {
5959
RPC_SOAK("sends 'soak_iterations' large_unary rpcs in a loop, each on the same channel"),
6060
CHANNEL_SOAK("sends 'soak_iterations' large_unary rpcs in a loop, each on a new channel"),
6161
ORCA_PER_RPC("report backend metrics per query"),
62-
ORCA_OOB("report backend metrics out-of-band");
62+
ORCA_OOB("report backend metrics out-of-band"),
63+
MAX_CONCURRENT_STREAMS_CONNECTION_SCALING("max concurrent streaming connection scaling");
6364

6465
private final String description;
6566

‎interop-testing/src/main/java/io/grpc/testing/integration/TestServiceClient.java‎

Lines changed: 93 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -563,7 +563,11 @@ private void runTest(TestCases testCase) throws Exception {
563563
tester.testOrcaOob();
564564
break;
565565
}
566-
566+
567+
case MAX_CONCURRENT_STREAMS_CONNECTION_SCALING: {
568+
tester.testMcs();
569+
break;
570+
}
567571
default:
568572
throw new IllegalArgumentException("Unknown test case: " + testCase);
569573
}
@@ -596,6 +600,7 @@ private ClientInterceptor maybeCreateAdditionalMetadataInterceptor(
596600
}
597601

598602
private class Tester extends AbstractInteropTest {
603+
599604
@Override
600605
protected ManagedChannelBuilder<?> createChannelBuilder() {
601606
boolean useGeneric = false;
@@ -979,31 +984,17 @@ public void testOrcaOob() throws Exception {
979984
.build();
980985

981986
final int retryLimit = 5;
982-
BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
983987
final Object lastItem = new Object();
988+
StreamingOutputCallResponseObserver streamingOutputCallResponseObserver =
989+
new StreamingOutputCallResponseObserver(lastItem);
984990
StreamObserver<StreamingOutputCallRequest> streamObserver =
985-
asyncStub.fullDuplexCall(new StreamObserver<StreamingOutputCallResponse>() {
986-
987-
@Override
988-
public void onNext(StreamingOutputCallResponse value) {
989-
queue.add(value);
990-
}
991-
992-
@Override
993-
public void onError(Throwable t) {
994-
queue.add(t);
995-
}
996-
997-
@Override
998-
public void onCompleted() {
999-
queue.add(lastItem);
1000-
}
1001-
});
991+
asyncStub.fullDuplexCall(streamingOutputCallResponseObserver);
1002992

1003993
streamObserver.onNext(StreamingOutputCallRequest.newBuilder()
1004994
.setOrcaOobReport(answer)
1005995
.addResponseParameters(ResponseParameters.newBuilder().setSize(1).build()).build());
1006-
assertThat(queue.take()).isInstanceOf(StreamingOutputCallResponse.class);
996+
assertThat(streamingOutputCallResponseObserver.take())
997+
.isInstanceOf(StreamingOutputCallResponse.class);
1007998
int i = 0;
1008999
for (; i < retryLimit; i++) {
10091000
Thread.sleep(1000);
@@ -1016,7 +1007,8 @@ public void onCompleted() {
10161007
streamObserver.onNext(StreamingOutputCallRequest.newBuilder()
10171008
.setOrcaOobReport(answer2)
10181009
.addResponseParameters(ResponseParameters.newBuilder().setSize(1).build()).build());
1019-
assertThat(queue.take()).isInstanceOf(StreamingOutputCallResponse.class);
1010+
assertThat(streamingOutputCallResponseObserver.take())
1011+
.isInstanceOf(StreamingOutputCallResponse.class);
10201012

10211013
for (i = 0; i < retryLimit; i++) {
10221014
Thread.sleep(1000);
@@ -1027,7 +1019,7 @@ public void onCompleted() {
10271019
}
10281020
assertThat(i).isLessThan(retryLimit);
10291021
streamObserver.onCompleted();
1030-
assertThat(queue.take()).isSameInstanceAs(lastItem);
1022+
assertThat(streamingOutputCallResponseObserver.take()).isSameInstanceAs(lastItem);
10311023
}
10321024

10331025
@Override
@@ -1054,6 +1046,85 @@ protected ServerBuilder<?> getHandshakerServerBuilder() {
10541046
protected int operationTimeoutMillis() {
10551047
return 15000;
10561048
}
1049+
1050+
class StreamingOutputCallResponseObserver implements
1051+
StreamObserver<StreamingOutputCallResponse> {
1052+
private final Object lastItem;
1053+
private final BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
1054+
1055+
public StreamingOutputCallResponseObserver(Object lastItem) {
1056+
this.lastItem = lastItem;
1057+
}
1058+
1059+
@Override
1060+
public void onNext(StreamingOutputCallResponse value) {
1061+
queue.add(value);
1062+
}
1063+
1064+
@Override
1065+
public void onError(Throwable t) {
1066+
queue.add(t);
1067+
}
1068+
1069+
@Override
1070+
public void onCompleted() {
1071+
queue.add(lastItem);
1072+
}
1073+
1074+
Object take() throws InterruptedException {
1075+
return queue.take();
1076+
}
1077+
}
1078+
1079+
public void testMcs() throws Exception {
1080+
final Object lastItem = new Object();
1081+
StreamingOutputCallResponseObserver responseObserver1 =
1082+
new StreamingOutputCallResponseObserver(lastItem);
1083+
StreamObserver<StreamingOutputCallRequest> streamObserver1 =
1084+
asyncStub.fullDuplexCall(responseObserver1);
1085+
StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
1086+
.addResponseParameters(ResponseParameters.newBuilder()
1087+
.setFillPeerSocketAddress(
1088+
Messages.BoolValue.newBuilder().setValue(true).build())
1089+
.build())
1090+
.build();
1091+
streamObserver1.onNext(request);
1092+
Object responseObj = responseObserver1.take();
1093+
StreamingOutputCallResponse callResponse = (StreamingOutputCallResponse) responseObj;
1094+
String clientSocketAddressInCall1 = callResponse.getPeerSocketAddress();
1095+
assertThat(clientSocketAddressInCall1).isNotEmpty();
1096+
1097+
StreamingOutputCallResponseObserver responseObserver2 =
1098+
new StreamingOutputCallResponseObserver(lastItem);
1099+
StreamObserver<StreamingOutputCallRequest> streamObserver2 =
1100+
asyncStub.fullDuplexCall(responseObserver2);
1101+
streamObserver2.onNext(request);
1102+
callResponse = (StreamingOutputCallResponse) responseObserver2.take();
1103+
String clientSocketAddressInCall2 = callResponse.getPeerSocketAddress();
1104+
1105+
assertThat(clientSocketAddressInCall1).isEqualTo(clientSocketAddressInCall2);
1106+
1107+
// The first connection is at max rpc call count of 2, so the 3rd rpc will cause a new
1108+
// connection to be created in the same subchannel and not get queued.
1109+
StreamingOutputCallResponseObserver responseObserver3 =
1110+
new StreamingOutputCallResponseObserver(lastItem);
1111+
StreamObserver<StreamingOutputCallRequest> streamObserver3 =
1112+
asyncStub.fullDuplexCall(responseObserver3);
1113+
streamObserver3.onNext(request);
1114+
callResponse = (StreamingOutputCallResponse) responseObserver3.take();
1115+
String clientSocketAddressInCall3 = callResponse.getPeerSocketAddress();
1116+
1117+
// This assertion is currently failing because connection scaling when MCS limit has been
1118+
// reached is not yet implemented in gRPC Java.
1119+
assertThat(clientSocketAddressInCall3).isNotEqualTo(clientSocketAddressInCall1);
1120+
1121+
streamObserver1.onCompleted();
1122+
streamObserver2.onCompleted();
1123+
streamObserver3.onCompleted();
1124+
assertThat(responseObserver1.take()).isSameInstanceAs(lastItem);
1125+
assertThat(responseObserver2.take()).isSameInstanceAs(lastItem);
1126+
assertThat(responseObserver3.take()).isSameInstanceAs(lastItem);
1127+
}
10571128
}
10581129

10591130
private static String validTestCasesHelpText() {

‎interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java‎

Lines changed: 47 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,18 @@
1616

1717
package io.grpc.testing.integration;
1818

19+
import static io.grpc.Grpc.TRANSPORT_ATTR_REMOTE_ADDR;
20+
1921
import com.google.common.base.Preconditions;
2022
import com.google.common.collect.Queues;
2123
import com.google.errorprone.annotations.concurrent.GuardedBy;
2224
import com.google.protobuf.ByteString;
25+
import io.grpc.Context;
26+
import io.grpc.Contexts;
2327
import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
2428
import io.grpc.Metadata;
2529
import io.grpc.ServerCall;
30+
import io.grpc.ServerCall.Listener;
2631
import io.grpc.ServerCallHandler;
2732
import io.grpc.ServerInterceptor;
2833
import io.grpc.Status;
@@ -42,6 +47,7 @@
4247
import io.grpc.testing.integration.Messages.StreamingOutputCallResponse;
4348
import io.grpc.testing.integration.Messages.TestOrcaReport;
4449
import io.grpc.testing.integration.TestServiceGrpc.AsyncService;
50+
import java.net.SocketAddress;
4551
import java.util.ArrayDeque;
4652
import java.util.Arrays;
4753
import java.util.HashMap;
@@ -59,8 +65,8 @@
5965
* sent in response streams.
6066
*/
6167
public class TestServiceImpl implements io.grpc.BindableService, AsyncService {
68+
static final Context.Key<SocketAddress> PEER_ADDRESS_CONTEXT_KEY = Context.key("peer-address");
6269
private final Random random = new Random();
63-
6470
private final ScheduledExecutorService executor;
6571
private final ByteString compressableBuffer;
6672
private final MetricRecorder metricRecorder;
@@ -441,7 +447,13 @@ public Queue<Chunk> toChunkQueue(StreamingOutputCallRequest request) {
441447
Queue<Chunk> chunkQueue = new ArrayDeque<>();
442448
int offset = 0;
443449
for (ResponseParameters params : request.getResponseParametersList()) {
444-
chunkQueue.add(new Chunk(params.getIntervalUs(), offset, params.getSize()));
450+
String peerSocketAddress = null;
451+
if (params.getFillPeerSocketAddress().getValue()) {
452+
SocketAddress peerAddress = PEER_ADDRESS_CONTEXT_KEY.get();
453+
peerSocketAddress = peerAddress != null ? peerAddress.toString() : "";
454+
}
455+
chunkQueue.add(
456+
new Chunk(params.getIntervalUs(), offset, params.getSize(), peerSocketAddress));
445457

446458
// Increment the offset past this chunk. Buffer need to be circular.
447459
offset = (offset + params.getSize()) % compressableBuffer.size();
@@ -459,11 +471,17 @@ private class Chunk {
459471
private final int delayMicroseconds;
460472
private final int offset;
461473
private final int length;
474+
private final String peerSocketAddress;
462475

463476
public Chunk(int delayMicroseconds, int offset, int length) {
477+
this(delayMicroseconds, offset, length, null);
478+
}
479+
480+
public Chunk(int delayMicroseconds, int offset, int length, String peerSocketAddress) {
464481
this.delayMicroseconds = delayMicroseconds;
465482
this.offset = offset;
466483
this.length = length;
484+
this.peerSocketAddress = peerSocketAddress;
467485
}
468486

469487
/**
@@ -472,10 +490,15 @@ public Chunk(int delayMicroseconds, int offset, int length) {
472490
private StreamingOutputCallResponse toResponse() {
473491
StreamingOutputCallResponse.Builder responseBuilder =
474492
StreamingOutputCallResponse.newBuilder();
475-
ByteString payload = generatePayload(compressableBuffer, offset, length);
476-
responseBuilder.setPayload(
477-
Payload.newBuilder()
478-
.setBody(payload));
493+
if (length > 0) {
494+
ByteString payload = generatePayload(compressableBuffer, offset, length);
495+
responseBuilder.setPayload(
496+
Payload.newBuilder()
497+
.setBody(payload));
498+
}
499+
if (peerSocketAddress != null) {
500+
responseBuilder.setPeerSocketAddress(peerSocketAddress);
501+
}
479502
return responseBuilder.build();
480503
}
481504
}
@@ -505,7 +528,8 @@ public static List<ServerInterceptor> interceptors() {
505528
return Arrays.asList(
506529
echoRequestHeadersInterceptor(Util.METADATA_KEY),
507530
echoRequestMetadataInHeaders(Util.ECHO_INITIAL_METADATA_KEY),
508-
echoRequestMetadataInTrailers(Util.ECHO_TRAILING_METADATA_KEY));
531+
echoRequestMetadataInTrailers(Util.ECHO_TRAILING_METADATA_KEY),
532+
new AddPeerAddressToContextInterceptor());
509533
}
510534

511535
/**
@@ -540,6 +564,22 @@ public void close(Status status, Metadata trailers) {
540564
};
541565
}
542566

567+
static class AddPeerAddressToContextInterceptor implements ServerInterceptor {
568+
@Override
569+
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
570+
Metadata headers, ServerCallHandler<ReqT, RespT> next) {
571+
SocketAddress peerAddress = call.getAttributes().get(TRANSPORT_ATTR_REMOTE_ADDR);
572+
573+
// Create a new context with the peer address value
574+
Context newContext = Context.current().withValue(PEER_ADDRESS_CONTEXT_KEY, peerAddress);
575+
try {
576+
return Contexts.interceptCall(newContext, call, headers, next);
577+
} catch (Exception ex) {
578+
throw new RuntimeException(ex);
579+
}
580+
}
581+
}
582+
543583
/**
544584
* Echoes request headers with the specified key from a client into response headers only.
545585
*/

‎interop-testing/src/main/java/io/grpc/testing/integration/TestServiceServer.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ public void run() {
7575
private int port = 8080;
7676
private boolean useTls = true;
7777
private boolean useAlts = false;
78+
private int mcsLimit = -1;
7879

7980
private ScheduledExecutorService executor;
8081
private Server server;
@@ -118,6 +119,10 @@ void parseArgs(String[] args) {
118119
usage = true;
119120
break;
120121
}
122+
} else if ("max_concurrent_streams_limit".equals(key)) {
123+
mcsLimit = Integer.parseInt(value);
124+
// TODO: Make Netty server builder usable for IPV6 as well (not limited to MCS handling)
125+
addressType = Util.AddressType.IPV4; // To use NettyServerBuilder
121126
} else {
122127
System.err.println("Unknown argument: " + key);
123128
usage = true;
@@ -141,6 +146,8 @@ void parseArgs(String[] args) {
141146
+ "\n for testing. Only effective when --use_alts=true."
142147
+ "\n --address_type=IPV4|IPV6|IPV4_IPV6"
143148
+ "\n What type of addresses to listen on. Default IPV4_IPV6"
149+
+ "\n --max_concurrent_streams_limit=LIMIT"
150+
+ "\n Set the maximum concurrent streams limit"
144151
);
145152
System.exit(1);
146153
}
@@ -186,6 +193,9 @@ void start() throws Exception {
186193
if (v4Address != null && !v4Address.equals(localV4Address)) {
187194
((NettyServerBuilder) serverBuilder).addListenAddress(v4Address);
188195
}
196+
if (mcsLimit != -1) {
197+
((NettyServerBuilder) serverBuilder).maxConcurrentCallsPerConnection(mcsLimit);
198+
}
189199
break;
190200
case IPV6:
191201
List<SocketAddress> v6Addresses = Util.getV6Addresses(port);

‎interop-testing/src/main/proto/grpc/testing/messages.proto‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,10 @@ message ResponseParameters {
159159
// implement the full compression tests by introspecting the call to verify
160160
// the response's compression status.
161161
BoolValue compressed = 3;
162+
163+
// Whether to request the server to send the requesting peer's socket
164+
// address in the response.
165+
BoolValue fill_peer_socket_address = 4;
162166
}
163167

164168
// Server-streaming request.
@@ -186,6 +190,9 @@ message StreamingOutputCallRequest {
186190
message StreamingOutputCallResponse {
187191
// Payload to increase response size.
188192
Payload payload = 1;
193+
194+
// The peer's socket address if requested.
195+
string peer_socket_address = 2;
189196
}
190197

191198
// For reconnect interop test only.

‎interop-testing/src/test/java/io/grpc/testing/integration/TestCasesTest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,8 @@ public void testCaseNamesShouldMapToEnums() {
6767
"cancel_after_first_response",
6868
"timeout_on_sleeping_server",
6969
"orca_per_rpc",
70-
"orca_oob"
70+
"orca_oob",
71+
"max_concurrent_streams_connection_scaling",
7172
};
7273

7374
// additional test cases

0 commit comments

Comments
 (0)