Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

core: allow per-service/method executor #8266

Merged
merged 4 commits into from Jun 21, 2021

Conversation

YifeiZhuang
Copy link
Contributor

@YifeiZhuang YifeiZhuang commented Jun 16, 2021

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

@YifeiZhuang YifeiZhuang requested review from ejona86 and removed request for ejona86 Jun 16, 2021
}

((SerializingExecutor)defaultExecutor).setExecutor(switchingExecutor);
defaultExecutor.execute(new CallHandled());
Copy link
Member

@ejona86 ejona86 Jun 17, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Contributor Author

@YifeiZhuang YifeiZhuang Jun 17, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Member

@ejona86 ejona86 Jun 17, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).

Copy link
Member

@ejona86 ejona86 left a comment

Looks like this approach will work reasonably well.

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

@ejona86 ejona86 Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Contributor Author

@YifeiZhuang YifeiZhuang Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@@ -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()) {
Copy link
Member

@ejona86 ejona86 Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Contributor Author

@YifeiZhuang YifeiZhuang Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

Copy link
Member

@ejona86 ejona86 Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@@ -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;
Copy link
Member

@ejona86 ejona86 Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Contributor Author

@YifeiZhuang YifeiZhuang Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Member

@ejona86 ejona86 Jun 18, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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
Copy link
Contributor Author

@YifeiZhuang YifeiZhuang commented Jun 18, 2021

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.

Copy link
Member

@ejona86 ejona86 left a comment

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

import javax.annotation.Nullable;

/**
* Defines what executor the server call is handled, based on each RPC call information at runtime.
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"handles the server call"

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

public void setExecutor(Executor executor) {
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@@ -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;
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s/wrapExecutor/wrappedExecutor/ to keep the name?

// 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
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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


private final class ServerCallParameters<ReqT, RespT> {
ServerCallImpl<ReqT, RespT> call;
ServerMethodDefinition<ReqT, RespT> methodDef;
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
}

final class ServerCallHandled extends ContextRunnable {
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

private <WReqT, WRespT> ServerStreamListener startWrappedCall(
String fullMethodName,
Copy link
Member

@ejona86 ejona86 Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).

Copy link
Contributor Author

@YifeiZhuang YifeiZhuang Jun 21, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@YifeiZhuang YifeiZhuang merged commit c540229 into grpc:master Jun 21, 2021
14 checks passed
YifeiZhuang added a commit to YifeiZhuang/grpc-java that referenced this issue Jun 25, 2021
YifeiZhuang added a commit that referenced this issue Jun 25, 2021
* Revert "core: fix boq conformance failures (#8281)"

This reverts commit c1ad5de.

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

This reverts commit c540229.
YifeiZhuang added a commit to YifeiZhuang/grpc-java that referenced this issue Jul 25, 2021
* 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.
@github-actions github-actions bot locked as resolved and limited conversation to collaborators Sep 20, 2021
Sign up for free to subscribe to this conversation on GitHub. Already have an account? Sign in.
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

None yet

2 participants