core: Reduce per-stream idle memory by 20%

Metadata was accidentally being retained after the start of the call.
That can be an overwhelming percentage of memory for an idle RPC; don't
do that. The other changes are considerably smaller, but I happened to
notice them and the changes are straight-forward without magic numbers
(e.g., there's many arrays that could be tuned).

The regular interop server uses 4600 bytes per full duplex stream while
idle, but much of that is Census recorded events hanging around. Keeping
the Census integration but removing the Census impl (so a noop is used)
drops that to 3000 bytes. This change brings that down to ~2450 bytes
(which is still including stuff from TestServiceImpl). But there's very
little Metadata in the interop tests, so absolute real-life savings
would be much higher (but relative real-life savings may be lower,
because the application will often have more state).

The measurements were captured using a modified
timeout_on_sleeping_server client that had 100,000 concurrent full
duplex calls on one connection.

Author: ejona86

core/src/main/java/io/grpc/internal/MessageDeframer.java

@@ -314,6 +314,12 @@ private boolean readRequiredBytes() {
     int totalBytesRead = 0;
     int deflatedBytesRead = 0;
     try {
+      // Avoid allocating nextFrame when idle
+      if (requiredLength > 0 && fullStreamDecompressor == null
+          && unprocessed.readableBytes() == 0) {
+        return false;
+      }
+
       if (nextFrame == null) {
         nextFrame = new CompositeReadableBuffer();
       }

core/src/main/java/io/grpc/internal/ServerImpl.java

@@ -151,7 +151,9 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
         InternalLogId.allocate("Server", String.valueOf(getListenSocketsIgnoringLifecycle()));
     // Fork from the passed in context so that it does not propagate cancellation, it only
     // inherits values.
-    this.rootContext = Preconditions.checkNotNull(rootContext, "rootContext").fork();
+    this.rootContext = Preconditions.checkNotNull(rootContext, "rootContext")
+        .fork()
+        .withValue(io.grpc.InternalServer.SERVER_CONTEXT_KEY, ServerImpl.this);
     this.decompressorRegistry = builder.decompressorRegistry;
     this.compressorRegistry = builder.compressorRegistry;
     this.transportFilters = Collections.unmodifiableList(
@@ -622,19 +624,7 @@ private void runInternal() {
           // An extremely short deadline may expire before stream.setListener(jumpListener).
           // This causes NPE as in issue: https://github.com/grpc/grpc-java/issues/6300
           // Delay of setting cancellationListener to context will fix the issue.
-          final class ServerStreamCancellationListener implements Context.CancellationListener {
-            @Override
-            public void cancelled(Context context) {
-              Status status = statusFromCancelled(context);
-              if (DEADLINE_EXCEEDED.getCode().equals(status.getCode())) {
-                // This should rarely get run, since the client will likely cancel the stream
-                // before the timeout is reached.
-                stream.cancel(status);
-              }
-            }
-          }
-
-          context.addListener(new ServerStreamCancellationListener(), directExecutor());
+          context.addListener(new ServerStreamCancellationListener(stream), directExecutor());
         }
       }
 
@@ -648,8 +638,7 @@ private Context.CancellableContext createContext(
 
       Context baseContext =
           statsTraceCtx
-              .serverFilterContext(rootContext)
-              .withValue(io.grpc.InternalServer.SERVER_CONTEXT_KEY, ServerImpl.this);
+              .serverFilterContext(rootContext);
 
       if (timeoutNanos == null) {
         return baseContext.withCancellation();
@@ -707,6 +696,31 @@ private <WReqT, WRespT> ServerStreamListener startWrappedCall(
     }
   }
 
+  /**
+   * Propagates context cancellation to the ServerStream.
+   *
+   * <p>This is outside of HandleServerCall because that class holds Metadata and other state needed
+   * only when starting the RPC. The cancellation listener will live for the life of the call, so we
+   * avoid that useless state being retained.
+   */
+  static final class ServerStreamCancellationListener implements Context.CancellationListener {
+    private final ServerStream stream;
+
+    ServerStreamCancellationListener(ServerStream stream) {
+      this.stream = checkNotNull(stream, "stream");
+    }
+
+    @Override
+    public void cancelled(Context context) {
+      Status status = statusFromCancelled(context);
+      if (DEADLINE_EXCEEDED.getCode().equals(status.getCode())) {
+        // This should rarely get run, since the client will likely cancel the stream
+        // before the timeout is reached.
+        stream.cancel(status);
+      }
+    }
+  }
+
   @Override
   public InternalLogId getLogId() {
     return logId;

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

@@ -45,12 +45,10 @@
 import java.util.ArrayDeque;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
 import java.util.Random;
-import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Semaphore;
@@ -511,27 +509,30 @@ public static List<ServerInterceptor> interceptors() {
   }
 
   /**
-   * Echo the request headers from a client into response headers and trailers. Useful for
+   * Echo a request header from a client into response headers and trailers. Useful for
    * testing end-to-end metadata propagation.
    */
-  private static ServerInterceptor echoRequestHeadersInterceptor(final Metadata.Key<?>... keys) {
-    final Set<Metadata.Key<?>> keySet = new HashSet<>(Arrays.asList(keys));
+  private static <T> ServerInterceptor echoRequestHeadersInterceptor(final Metadata.Key<T> key) {
     return new ServerInterceptor() {
       @Override
       public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
           ServerCall<ReqT, RespT> call,
-          final Metadata requestHeaders,
+          Metadata requestHeaders,
           ServerCallHandler<ReqT, RespT> next) {
+        if (!requestHeaders.containsKey(key)) {
+          return next.startCall(call, requestHeaders);
+        }
+        T value = requestHeaders.get(key);
         return next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
               @Override
               public void sendHeaders(Metadata responseHeaders) {
-                responseHeaders.merge(requestHeaders, keySet);
+                responseHeaders.put(key, value);
                 super.sendHeaders(responseHeaders);
               }
 
               @Override
               public void close(Status status, Metadata trailers) {
-                trailers.merge(requestHeaders, keySet);
+                trailers.put(key, value);
                 super.close(status, trailers);
               }
             }, requestHeaders);
@@ -540,52 +541,48 @@ public void close(Status status, Metadata trailers) {
   }
 
   /**
-   * Echoes request headers with the specified key(s) from a client into response headers only.
+   * Echoes request headers with the specified key from a client into response headers only.
    */
-  private static ServerInterceptor echoRequestMetadataInHeaders(final Metadata.Key<?>... keys) {
-    final Set<Metadata.Key<?>> keySet = new HashSet<>(Arrays.asList(keys));
+  private static <T> ServerInterceptor echoRequestMetadataInHeaders(final Metadata.Key<T> key) {
     return new ServerInterceptor() {
       @Override
       public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
           ServerCall<ReqT, RespT> call,
           final Metadata requestHeaders,
           ServerCallHandler<ReqT, RespT> next) {
+        if (!requestHeaders.containsKey(key)) {
+          return next.startCall(call, requestHeaders);
+        }
+        T value = requestHeaders.get(key);
         return next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
           @Override
           public void sendHeaders(Metadata responseHeaders) {
-            responseHeaders.merge(requestHeaders, keySet);
+            responseHeaders.put(key, value);
             super.sendHeaders(responseHeaders);
           }
-
-          @Override
-          public void close(Status status, Metadata trailers) {
-            super.close(status, trailers);
-          }
         }, requestHeaders);
       }
     };
   }
 
   /**
-   * Echoes request headers with the specified key(s) from a client into response trailers only.
+   * Echoes request headers with the specified key from a client into response trailers only.
    */
-  private static ServerInterceptor echoRequestMetadataInTrailers(final Metadata.Key<?>... keys) {
-    final Set<Metadata.Key<?>> keySet = new HashSet<>(Arrays.asList(keys));
+  private static <T> ServerInterceptor echoRequestMetadataInTrailers(final Metadata.Key<T> key) {
     return new ServerInterceptor() {
       @Override
       public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
           ServerCall<ReqT, RespT> call,
           final Metadata requestHeaders,
           ServerCallHandler<ReqT, RespT> next) {
+        if (!requestHeaders.containsKey(key)) {
+          return next.startCall(call, requestHeaders);
+        }
+        T value = requestHeaders.get(key);
         return next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
-          @Override
-          public void sendHeaders(Metadata responseHeaders) {
-            super.sendHeaders(responseHeaders);
-          }
-
           @Override
           public void close(Status status, Metadata trailers) {
-            trailers.merge(requestHeaders, keySet);
+            trailers.put(key, value);
             super.close(status, trailers);
           }
         }, requestHeaders);