core: allow per-service/method executor #8266

YifeiZhuang · 2021-06-16T22:21:37Z

For #7874
Technical strategy:

Split StreamCreated() contextRunnable into two when this feature is needed: one for method lookup, the other for server call handling. methodLookup() runs on default executor, handleServerCall() may run on the executorSupplier executor.
- callbacks are queued after methodLookup() and handleServerCall()
Make executor settable in serializing executor to switch executor for the server call handling runnable as a result of the outcome of the method lookup runnable.

@ejona86 can I get your early feedback

tracking experimental API #8274

ejona86 · 2021-06-17T14:26:47Z

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

+          }
+
+          ((SerializingExecutor)defaultExecutor).setExecutor(switchingExecutor);
+          defaultExecutor.execute(new CallHandled());


In our discussion I thought we decided this approach was broken because this Runnable might be scheduled behind other runnables that are going to use the listener. For example, if the client cancels the call before this line runs.

Although, I realize now a way to resolve that. We can unconditionally enqueue this Runnable at the very beginning.

wrappedExecutor.execute(new LookupMethod()); // creates Call and calls setExecutor() if necessary wrappedExecutor.execute(new StreamCreated()); // calls jumpListener.setListener()

If we want to optimize that, we avoid LookupMethod when unnecessary. Before we were trying to avoid enqueuing the later Runnables, but it seems much easier to avoid enqueuing the earlier runnables (if we care).

The main annoyance is we'll want to pass the created ServerCall to StreamCreated, but we'll have to pass it after the runnable is constructed.

In our discussion I thought we decided this approach was broken because this Runnable might be scheduled behind other runnables that are going to use the listener. For example, if the client cancels the call before this line runs.

It is broken if we execute Runnables that uses the listener immediately. So, we enqueue them all and deliver all at once. But this enqueue needs synchronization that's probably a problem.

Although, I realize now a way to resolve that. We can unconditionally enqueue this Runnable at the very beginning.

wrappedExecutor.execute(new LookupMethod()); // creates Call and calls setExecutor() if necessary wrappedExecutor.execute(new StreamCreated()); // calls jumpListener.setListener()

If we want to optimize that, we avoid LookupMethod when unnecessary. Before we were trying to avoid enqueuing the later Runnables, but it seems much easier to avoid enqueuing the earlier runnables (if we care).

Oh this approach sounds nice. i'll try it.
One optimization this approach can't do is to put the second runnable inline of the first one, unless we can cancel the second one before the first one finishes.

The main annoyance is we'll want to pass the created ServerCall to StreamCreated, but we'll have to pass it after the runnable is constructed.

It is broken if we execute Runnables that uses the listener immediately. So, we enqueue them all and deliver all at once.

Oh, I see it now. I should have looked at that more closely.

One optimization this approach can't do is to put the second runnable inline of the first one, unless we can cancel the second one before the first one finishes.

SerializingExecutor will do that naturally if the executor doesn't change. Yes, it will be a different Runnable instance, but they are guaranteed to run back-to-back on the same thread (assuming the executor doesn't change).

ejona86

Looks like this approach will work reasonably well.

ejona86 · 2021-06-18T14:38:50Z

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

+        public void runInContext() {
+          ServerStreamListener listener = NOOP_LISTENER;
+          ServerCallParameters<?,?> callParameters;
+          if (future.isCancelled()) {


This should be done within the try-finally, as even if cancelled we need to set the listener (to NOOP).

oh, because it was handled in the lookupMethod(), but yea I think it is better to put it in callHandled() try-finally

ejona86 · 2021-06-18T14:42:46Z

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

@@ -466,14 +479,14 @@ public void streamCreated(ServerStream stream, String methodName, Metadata heade

    private void streamCreatedInternal(
        final ServerStream stream, final String methodName, final Metadata headers, final Tag tag) {
-      final Executor wrappedExecutor;
+      final Executor defaultExecutor;
      // This is a performance optimization that avoids the synchronization and queuing overhead
      // that comes with SerializingExecutor.
      if (executor == directExecutor()) {


It seems like we'll need to disable these two optimizations if an executor supplier is provided. I think the threading for optimizeForDirectExecutor requires it to be done inline before this method returns. It is safe to call it even if direct executor isn't used, but if/when migrating deframer gets enabled again getting it wrong hurts the non-direct case much more than it helps the direct case; that is to say, if we don't know which executor will be used, it is better to error on the side of not calling the method.

But executor supplier is never used here when direct executor is enabled. It looks optimizeForDirectExecutor just means we should use SerializeReentrantCallsDirectExecutor, and that would run tasks inline but still queued if already executing, so, it is still safe when we would have two contextRunnables, because listener call backs would still be queued at SerializeReentrantCallsDirectExecutor.
How does migrating deframer hurt the non-direct case?
Did you mean we should not allow them to co-exist?
executor==directExecutor and executor_supplier != null validation error at server builder.
executor==directExecutor and executor_supplier == null: optimizeForDirectExecutor and use SerializeReentrantCallsDirectExecutor
executor!=directExecutor and executor_supplier != null: Just SerializingExecutor, no optimization.
executor!=directExecutor and executor_supplier == null: always use executor(I need to check and fix)

But executor supplier is never used here when direct executor is enabled.

But that is clearly not right/incomplete. The executor for lookups in the fallback registry may be direct executor and that does not seem to be cause to ignore a user's executor supplier. I was discussing how we'll go about mixing all this stuff together.

It looks optimizeForDirectExecutor just means we should use SerializeReentrantCallsDirectExecutor

SerializeReentrantCallsDirectExecutor is different from optimizeForDirectExecutor. optimizeForDirectExecutor was intended to disable MigratingDeframer when using direct executor, although MigratingDeframer is disabled for all cases at the moment.

How does migrating deframer hurt the non-direct case?

It helps the non-direct case. However, it adds overhead that can hurt the direct case, so the code was disabling it when direct executor was used. I was saying the overhead it adds to direct is pretty small compared to how much it helps the non-direct case; we can simply disable this optimization (i.e., not call optimizeForDirectExecutor) if an executor supplier is provided.

executor==directExecutor and executor_supplier != null validation error at server builder.

I see no reason to cause an error. The two conceptually can make sense. It is just a question of how many optimizations does it disable. I'm suggesting we'll treat this the same as executor!=directExecutor and executor_supplier != null.

ejona86 · 2021-06-18T14:46:16Z

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

@@ -59,7 +59,7 @@ private static AtomicHelper getAtomicHelper() {
  private static final int RUNNING = -1;

  /** Underlying executor that all submitted Runnable objects are run on. */
-  private final Executor executor;
+  private volatile Executor executor;


Note: we don't actually need this to be volatile.

oh is it because it is only set and execute from the same network thread(eventloop)?

It is only called from within the Runnable currently being executed (r.run();). That is not the network thread. While a Runnable is running, runState == RUNNING and the network thread will not access executor.

YifeiZhuang · 2021-06-18T16:37:43Z

Looks like this approach will work reasonably well.

Yeah! It is so simple, I like it. ;)
But then I have to think about API level organization, plus more tests.

ejona86

I think some of the comments and error strings are the main important parts to improve. Most other things are more minor.

ejona86 · 2021-06-21T14:11:58Z

api/src/main/java/io/grpc/ServerCallExecutorSupplier.java

+import javax.annotation.Nullable;
+
+/**
+ * Defines what executor the server call is handled, based on each RPC call information at runtime.


"handles the server call"

ejona86 · 2021-06-21T14:15:54Z

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

@@ -76,6 +76,11 @@ public SerializingExecutor(Executor executor) {
    this.executor = executor;
  }

+  public void setExecutor(Executor executor) {


Document that this can only be called from a Runnable running within the SerializingExecutor.

ejona86 · 2021-06-21T14:17:15Z

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

@@ -466,14 +469,14 @@ public void streamCreated(ServerStream stream, String methodName, Metadata heade

    private void streamCreatedInternal(
        final ServerStream stream, final String methodName, final Metadata headers, final Tag tag) {
-      final Executor wrappedExecutor;
+      final Executor wrapExecutor;


s/wrapExecutor/wrappedExecutor/ to keep the name?

ejona86 · 2021-06-21T14:22:03Z

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

+      // Run in serializing executor so jumpListener.setListener() is called before any callbacks
+      // are delivered, including any errors. Callbacks can still be triggered.
+      // If callExecutor needs no executor switch, they will be queued at serializing executor.
+      // If callExecutor needs a switch due to executorSupplier, they will be queued at jumpListener


This looks out-of-date. I guess the old comment isn't inaccurate now, although don't know if we want to describe more.

ejona86 · 2021-06-21T15:02:31Z

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

+
+    private final class ServerCallParameters<ReqT, RespT> {
+      ServerCallImpl<ReqT, RespT> call;
+      ServerMethodDefinition<ReqT, RespT> methodDef;


Use ServerCallHandler here to be more clear that the MethodDescriptor within the definition isn't used?

ejona86 · 2021-06-21T15:22:51Z

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

+            }
+            if (!future.isDone() || (callParameters = future.get()) == null) {
+              Status status = Status.INTERNAL.withDescription(
+                      "Fail to start server call.");


Make the message a bit more clear there's a bug in grpc.

ejona86 · 2021-06-21T15:34:32Z

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

+              return;
+            }
+            if (!future.isDone() || (callParameters = future.get()) == null) {
+              Status status = Status.INTERNAL.withDescription(


Maybe this should throw this as an exception? If it's an exception then it will be logged (and the catch will handle it). Seems that would make it more likely we'd learn about the bug.

ejona86 · 2021-06-21T15:35:37Z

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

+        }
+      }
+
+      final class ServerCallHandled extends ContextRunnable {


Why "handled" in the name, since this handles the RPC, it isn't run after the RPC is handled. Maybe HandleServerCall?

ejona86 · 2021-06-21T15:38:29Z

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

+                                  ServerMethodDefinition<ReqT, RespT> methodDef) {
+        this.call = call;
+        this.methodDef = methodDef;
+      }
    }

    private <WReqT, WRespT> ServerStreamListener startWrappedCall(
        String fullMethodName,


If you are interested: Looks like we can remove this argument by using params.call.getMethodDescriptor().getFullMethodName(). Looks like ServerCallHandled then wouldn't need to look at the fullMethodName at all (only MethodLookup would).

oh looks this is only for logging the original fullMethodName. And this is an existing code path.

This reverts commit c540229.

* Revert "core: fix boq conformance failures (#8281)" This reverts commit c1ad5de. * Revert "core: allow per-service/method executor (#8266)" This reverts commit c540229.

* Revert "core: fix boq conformance failures (grpc#8281)" This reverts commit c1ad5de. * Revert "core: allow per-service/method executor (grpc#8266)" This reverts commit c540229.

two context

8d5ddba

YifeiZhuang mentioned this pull request Jun 16, 2021

core: allow per-service/method executor #8257

Closed

YifeiZhuang requested review from ejona86 and removed request for ejona86 June 16, 2021 22:25

ejona86 reviewed Jun 17, 2021

View reviewed changes

schedule two runnable proactively

d1610ff

ejona86 reviewed Jun 18, 2021

View reviewed changes

add executor supplier

5ff50e9

ejona86 approved these changes Jun 21, 2021

View reviewed changes

fix comments, throw exception future not set

3d333cb

YifeiZhuang merged commit c540229 into grpc:master Jun 21, 2021

ejona86 mentioned this pull request Jun 24, 2021

Add the ability to have a separate thread pool/executor for each endpoint #7874

Closed

YifeiZhuang added a commit to YifeiZhuang/grpc-java that referenced this pull request Jun 25, 2021

Revert "core: allow per-service/method executor (grpc#8266)"

47a32bf

This reverts commit c540229.

github-actions bot locked as resolved and limited conversation to collaborators Sep 20, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

core: allow per-service/method executor #8266

core: allow per-service/method executor #8266

YifeiZhuang commented Jun 16, 2021 •

edited

ejona86 Jun 17, 2021

YifeiZhuang Jun 17, 2021

ejona86 Jun 17, 2021

ejona86 left a comment

ejona86 Jun 18, 2021

YifeiZhuang Jun 18, 2021

ejona86 Jun 18, 2021

YifeiZhuang Jun 18, 2021

ejona86 Jun 18, 2021

ejona86 Jun 18, 2021

YifeiZhuang Jun 18, 2021

ejona86 Jun 18, 2021

YifeiZhuang commented Jun 18, 2021

ejona86 left a comment

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

ejona86 Jun 21, 2021

YifeiZhuang Jun 21, 2021

core: allow per-service/method executor #8266

core: allow per-service/method executor #8266

Conversation

YifeiZhuang commented Jun 16, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ejona86 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YifeiZhuang commented Jun 18, 2021

ejona86 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YifeiZhuang commented Jun 16, 2021 •

edited