Server-side timeout mechanism

util/src/main/java/io/grpc/util/SerializingServerCall.java

@@ -0,0 +1,183 @@
+/*
+ * Copyright 2017 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.util;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
+import io.grpc.Attributes;
+import io.grpc.ExperimentalApi;
+import io.grpc.ForwardingServerCall;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.Status;
+import io.grpc.internal.SerializingExecutor;
+import java.util.concurrent.ExecutionException;
+import javax.annotation.Nullable;
+
+/**
+ * A {@link ServerCall} that wraps around a non thread safe delegate and provides thread safe
+ * access by serializing everything on an executor.
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/2189")
+class SerializingServerCall<ReqT, RespT> extends
+    ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> {
+  private static final String ERROR_MSG = "Encountered error during serialized access";
+  private final SerializingExecutor serializingExecutor =
+      new SerializingExecutor(MoreExecutors.directExecutor());
+  private boolean closeCalled = false;
+
+  SerializingServerCall(ServerCall<ReqT, RespT> delegate) {
+    super(delegate);
+  }
+
+  @Override
+  public void sendMessage(final RespT message) {
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        SerializingServerCall.super.sendMessage(message);
+      }
+    });
+  }
+
+  @Override
+  public void request(final int numMessages) {
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        SerializingServerCall.super.request(numMessages);
+      }
+    });
+  }
+
+  @Override
+  public void sendHeaders(final Metadata headers) {
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        SerializingServerCall.super.sendHeaders(headers);
+      }
+    });
+  }
+
+  @Override
+  public void close(final Status status, final Metadata trailers) {
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        if (!closeCalled) {
+          closeCalled = true;
+
+          SerializingServerCall.super.close(status, trailers);
+        }
+      }
+    });
+  }
+
+  @Override
+  public boolean isReady() {
+    final SettableFuture<Boolean> retVal = SettableFuture.create();
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        retVal.set(SerializingServerCall.super.isReady());
+      }
+    });
+    try {
+      return retVal.get();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    } catch (ExecutionException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    }
+  }
+
+  @Override
+  public boolean isCancelled() {
+    final SettableFuture<Boolean> retVal = SettableFuture.create();
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        retVal.set(SerializingServerCall.super.isCancelled());
+      }
+    });
+    try {
+      return retVal.get();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    } catch (ExecutionException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    }
+  }
+
+  @Override
+  public void setMessageCompression(final boolean enabled) {
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        SerializingServerCall.super.setMessageCompression(enabled);
+      }
+    });
+  }
+
+  @Override
+  public void setCompression(final String compressor) {
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        SerializingServerCall.super.setCompression(compressor);
+      }
+    });
+  }
+
+  @Override
+  public Attributes getAttributes() {
+    final SettableFuture<Attributes> retVal = SettableFuture.create();
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        retVal.set(SerializingServerCall.super.getAttributes());
+      }
+    });
+    try {
+      return retVal.get();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    } catch (ExecutionException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    }
+  }
+
+  @Nullable
+  @Override
+  public String getAuthority() {
+    final SettableFuture<String> retVal = SettableFuture.create();
+    serializingExecutor.execute(new Runnable() {
+      @Override
+      public void run() {
+        retVal.set(SerializingServerCall.super.getAuthority());
+      }
+    });
+    try {
+      return retVal.get();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    } catch (ExecutionException e) {
+      throw new RuntimeException(ERROR_MSG, e);
+    }
+  }
+}

util/src/main/java/io/grpc/util/ServerCallTimeoutInterceptor.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2023 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.util;
+
+import io.grpc.Context;
+import io.grpc.ExperimentalApi;
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+
+/**
+ * An optional ServerInterceptor that can interrupt server calls that are running for too long time.
+ * In this way, it prevents problematic code from using up all threads.
+ *
+ * <p>How to use: you can add it to your server using ServerBuilder#intercept(ServerInterceptor).
+ *
+ * <p>Limitation: it only applies the timeout to unary calls
+ * (streaming calls will still run without timeout).
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/10361")
+public class ServerCallTimeoutInterceptor implements ServerInterceptor {
+
+  private final ServerTimeoutManager serverTimeoutManager;
+
+  public ServerCallTimeoutInterceptor(ServerTimeoutManager serverTimeoutManager) {
+    this.serverTimeoutManager = serverTimeoutManager;
+  }
+
+  @Override
+  public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+      ServerCall<ReqT, RespT> serverCall,
+      Metadata metadata,
+      ServerCallHandler<ReqT, RespT> serverCallHandler) {
+    // Only intercepts unary calls because the timeout is inapplicable to streaming calls.
+    if (serverCall.getMethodDescriptor().getType().clientSendsOneMessage()) {
+      ServerCall<ReqT, RespT> serializingServerCall = new SerializingServerCall<>(serverCall);
+      Context.CancellableContext timeoutContext =
+              serverTimeoutManager.startTimeoutContext(serializingServerCall);
+      if (timeoutContext != null) {
+        return new TimeoutServerCallListener<>(
+                serverCallHandler.startCall(serializingServerCall, metadata),
+                timeoutContext,
+                serverTimeoutManager);
+      }
+    }
+    return serverCallHandler.startCall(serverCall, metadata);
+  }
+
+  /** A listener that intercepts RPC callbacks for timeout control. */
+  static class TimeoutServerCallListener<ReqT>
+      extends ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT> {
+
+    private final Context.CancellableContext context;
+    private final ServerTimeoutManager serverTimeoutManager;
+
+    private TimeoutServerCallListener(
+        ServerCall.Listener<ReqT> delegate,
+        Context.CancellableContext context,
+        ServerTimeoutManager serverTimeoutManager) {
+      super(delegate);
+      this.context = context;
+      this.serverTimeoutManager = serverTimeoutManager;
+    }
+
+    @Override
+    public void onMessage(ReqT message) {
+      Context previous = context.attach();
+      try {
+        super.onMessage(message);
+      } finally {
+        context.detach(previous);
+      }
+    }
+
+    /**
+     * Intercepts onHalfClose() because the application RPC method is called in it. See
+     * io.grpc.stub.ServerCalls.UnaryServerCallHandler.UnaryServerCallListener
+     */
+    @Override
+    public void onHalfClose() {
+      serverTimeoutManager.withInterruption(context, super::onHalfClose);
+    }
+
+    @Override
+    public void onCancel() {
+      Context previous = context.attach();
+      try {
+        super.onCancel();
+      } finally {
+        context.detach(previous);
+      }
+    }
+
+    @Override
+    public void onComplete() {
+      Context previous = context.attach();
+      try {
+        super.onComplete();
+      } finally {
+        context.detach(previous);
+      }
+    }
+
+    @Override
+    public void onReady() {
+      Context previous = context.attach();
+      try {
+        super.onReady();
+      } finally {
+        context.detach(previous);
+      }
+    }
+  }
+}