Absolute deadline propagation grpc#11966

Author: kannanjgithub

examples/build.gradle

@@ -29,6 +29,7 @@ dependencies {
     implementation "io.grpc:grpc-protobuf:${grpcVersion}"
     implementation "io.grpc:grpc-services:${grpcVersion}"
     implementation "io.grpc:grpc-stub:${grpcVersion}"
+    implementation "io.grpc:grpc-core:${grpcVersion}"
     compileOnly "org.apache.tomcat:annotations-api:6.0.53"
 
     // examples/advanced need this for JsonFormat
@@ -45,7 +46,7 @@ dependencies {
 protobuf {
     protoc { artifact = "com.google.protobuf:protoc:${protocVersion}" }
     plugins {
-        grpc { artifact = "io.grpc:protoc-gen-grpc-java:${grpcVersion}" }
+        grpc { artifact = "io.grpc:protoc-gen-grpc-java:1.70.0" }
     }
     generateProtoTasks {
         all()*.plugins { grpc {} }

examples/src/main/java/io/grpc/examples/deadline/AbsoluteDeadlineServerStreamTracerFactory.java

@@ -0,0 +1,31 @@
+package io.grpc.examples.deadline;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.Context;
+import io.grpc.Metadata;
+import io.grpc.ServerStreamTracer;
+import io.grpc.internal.GrpcUtil;
+
+class AbsoluteDeadlineServerStreamTracerFactory extends ServerStreamTracer.Factory {
+  public static final Metadata.Key<String> GRPC_DEADLINE = Metadata.Key.of("grpc-deadline",
+          Metadata.ASCII_STRING_MARSHALLER);
+  public static final ServerStreamTracer NOOP_STREAM_TRACER = new ServerStreamTracer() {
+  };
+  private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+
+  @Override
+  public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
+
+    headers.discardAll(GrpcUtil.TIMEOUT_KEY);
+
+    return new ServerStreamTracer() {
+      @Override
+      public Context filterContext(Context context) {
+        return context.withDeadlineAfter(300, TimeUnit.MILLISECONDS, scheduler);
+      }
+    };
+  }
+}

examples/src/main/java/io/grpc/examples/deadline/DeadlineClient.java

@@ -79,30 +79,10 @@ public static void main(String[] args) throws Exception {
     try {
       DeadlineClient client = new DeadlineClient(channel);
 
-      // The server takes 500ms to process the call, so setting a deadline further in the future we
-      // should get a successful response.
-      logger.info("Calling server with a generous deadline, expected to work");
-      client.greet("deadline client", 1000);
-
       // A smaller deadline will result in us getting a DEADLINE_EXCEEDED error.
       logger.info(
           "Calling server with an unrealistic (300ms) deadline, expecting a DEADLINE_EXCEEDED");
       client.greet("deadline client", 300);
-
-      // Including the "propagate" magic string in the request will cause the server to call itself
-      // to simulate a situation where a server needs to call another server to satisfy the original
-      // request. This will double the time it takes to respond to the client request, but with
-      // an increased deadline we should get a successful response.
-      logger.info("Calling server with propagation and a generous deadline, expected to work");
-      client.greet("deadline client [propagate]", 2000);
-
-      // With this propagated call we reduce the deadline making it impossible for the both the
-      // first server call and the propagated one to succeed. You should see the call fail with
-      // DEADLINE_EXCEEDED, and you should also see DEADLINE_EXCEEDED in the server output as it
-      // runs out of time waiting for the propagated call to finish.
-      logger.info(
-          "Calling server with propagation and a generous deadline, expecting a DEADLINE_EXCEEDED");
-      client.greet("deadline client [propagate]", 1000);
     } finally {
       channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
     }

examples/src/main/java/io/grpc/examples/deadline/DeadlineServer.java

@@ -20,11 +20,14 @@
 import io.grpc.InsecureChannelCredentials;
 import io.grpc.InsecureServerCredentials;
 import io.grpc.Server;
+import io.grpc.ServerInterceptors;
 import io.grpc.examples.helloworld.GreeterGrpc;
 import io.grpc.examples.helloworld.HelloReply;
 import io.grpc.examples.helloworld.HelloRequest;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
 
@@ -38,6 +41,9 @@ private void start() throws IOException {
     int port = 50051;
     SlowGreeter slowGreeter = new SlowGreeter();
     server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
+        /*.addService(ServerInterceptors.intercept(slowGreeter, new DeadlineServerInterceptor(
+                Executors.newScheduledThreadPool(1))))*/
+        .addStreamTracerFactory(new AbsoluteDeadlineServerStreamTracerFactory())
         .addService(slowGreeter)
         .build()
         .start();

examples/src/main/java/io/grpc/examples/deadline/DeadlineServerInterceptor.java

@@ -0,0 +1,45 @@
+package io.grpc.examples.deadline;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.Context;
+import io.grpc.Contexts;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+import static io.grpc.Contexts.statusFromCancelled;
+import static io.grpc.Status.DEADLINE_EXCEEDED;
+
+class DeadlineServerInterceptor implements ServerInterceptor {
+  public static final Metadata.Key<String> GRPC_DEADLINE = Metadata.Key.of("grpc-deadline",
+          Metadata.ASCII_STRING_MARSHALLER);
+
+  private final ScheduledExecutorService scheduler;
+
+  public DeadlineServerInterceptor(ScheduledExecutorService scheduler) {
+    this.scheduler = scheduler;
+  }
+
+  @Override
+  public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
+                                                               ServerCallHandler<ReqT, RespT> next) {
+
+    Context.CancellableContext context = Context.current()
+            .fork()
+            .withDeadlineAfter(100, TimeUnit.MILLISECONDS, scheduler);
+
+    context.addListener(ctx -> {
+      Status status = statusFromCancelled(ctx);
+      if (DEADLINE_EXCEEDED.getCode().equals(status.getCode())) {
+        call.close(status, new Metadata());
+      }
+    }, directExecutor());
+
+    return Contexts.interceptCall(context, call, headers, next);
+  }
+}