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.

Sign up for GitHub

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

Jump to bottom

core: allow per-service/method executor #8266

Merged
merged 4 commits into from Jun 21, 2021
Merged

core: allow per-service/method executor #8266

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8d5ddba
two context
YifeiZhuang Jun 16, 2021
d1610ff
schedule two runnable proactively
YifeiZhuang Jun 17, 2021
5ff50e9
add executor supplier
YifeiZhuang Jun 18, 2021
3d333cb
fix comments, throw exception future not set
YifeiZhuang Jun 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
10 changes: 8 additions & 2 deletions core/src/main/java/io/grpc/internal/SerializingExecutor.java
View file
Open in desktop
Expand Up @@ -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

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

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

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.


/** A list of Runnables to be run in order. */
private final Queue<Runnable> runQueue = new ConcurrentLinkedQueue<>();
Expand All @@ -76,6 +76,11 @@ public SerializingExecutor(Executor executor) {
this.executor = executor;
}

public void setExecutor(Executor executor) {
Copy link
Member

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.

Preconditions.checkNotNull(executor, "'executor' must not be null.");
this.executor = executor;
}

/**
* Runs the given runnable strictly after all Runnables that were submitted
* before it, and using the {@code executor} passed to the constructor. .
Expand Down Expand Up @@ -118,7 +123,8 @@ private void schedule(@Nullable Runnable removable) {
public void run() {
Runnable r;
try {
while ((r = runQueue.poll()) != null) {
Executor oldExecutor = executor;
while ((r = runQueue.poll()) != null && oldExecutor == executor ) {
try {
r.run();
} catch (RuntimeException e) {
Expand Down
153 changes: 112 additions & 41 deletions core/src/main/java/io/grpc/internal/ServerImpl.java
View file
Open in desktop
Expand Up @@ -125,6 +125,7 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
private final InternalChannelz channelz;
private final CallTracer serverCallTracer;
private final Deadline.Ticker ticker;
private final ExecutorSupplier executorSupplier;

/**
* Construct a server.
Expand Down Expand Up @@ -159,6 +160,18 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
this.serverCallTracer = builder.callTracerFactory.create();
this.ticker = checkNotNull(builder.ticker, "ticker");
channelz.addServer(this);
this.executorSupplier = new ExecutorSupplier() {
@Override
public Executor getExecutor(ServerCall<?, ?> call,
Metadata header) {
return executor;
}
};
}


public interface ExecutorSupplier {
Executor getExecutor(ServerCall<?,?> call, Metadata metadata);
}

/**
Expand Down Expand Up @@ -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

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

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

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.

wrappedExecutor = new SerializeReentrantCallsDirectExecutor();
defaultExecutor = new SerializeReentrantCallsDirectExecutor();
stream.optimizeForDirectExecutor();
} else {
wrappedExecutor = new SerializingExecutor(executor);
defaultExecutor = new SerializingExecutor(executor);
}

if (headers.containsKey(MESSAGE_ENCODING_KEY)) {
Expand All @@ -499,14 +512,17 @@ private void streamCreatedInternal(

final JumpToApplicationThreadServerStreamListener jumpListener
= new JumpToApplicationThreadServerStreamListener(
wrappedExecutor, executor, stream, context, tag);
defaultExecutor, executor, stream, context, tag);
stream.setListener(jumpListener);
// Run in wrappedExecutor so jumpListener.setListener() is called before any callbacks
// are delivered, including any errors. Callbacks can still be triggered, but they will be
// queued.

final class StreamCreated extends ContextRunnable {
StreamCreated() {
final SettableFuture<ServerCallParameters<?,?>> future = SettableFuture.create();
// 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

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.

// first then delivered to serializing executor for the second queueing.

final class MethodLookup extends ContextRunnable {
MethodLookup() {
super(context);
}

Expand All @@ -522,31 +538,36 @@ public void runInContext() {
}

private void runInternal() {
ServerStreamListener listener = NOOP_LISTENER;
ServerMethodDefinition<?, ?> wrapMethod;
ServerCallParameters<?, ?> callParams;
try {
ServerMethodDefinition<?, ?> method = registry.lookupMethod(methodName);
if (method == null) {
method = fallbackRegistry.lookupMethod(methodName, stream.getAuthority());
}
if (method == null) {
Status status = Status.UNIMPLEMENTED.withDescription(
"Method not found: " + methodName);
"Method not found: " + methodName);
// TODO(zhangkun83): this error may be recorded by the tracer, and if it's kept in
// memory as a map whose key is the method name, this would allow a misbehaving
// client to blow up the server in-memory stats storage by sending large number of
// distinct unimplemented method
// names. (https://github.com/grpc/grpc-java/issues/2285)
stream.close(status, new Metadata());
context.cancel(null);
jumpListener.setListener(NOOP_LISTENER);
future.cancel(false);
return;
}
listener = startCall(stream, methodName, method, headers, context, statsTraceCtx, tag);
wrapMethod = wrapMethod(stream, method, statsTraceCtx);
callParams = maySwitchExecutor(wrapMethod, stream, headers, context, tag);
future.set(callParams);
} catch (Throwable t) {
stream.close(Status.fromThrowable(t), new Metadata());
context.cancel(null);
future.cancel(false);
jumpListener.setListener(NOOP_LISTENER);
throw t;
} finally {
jumpListener.setListener(listener);
}

// An extremely short deadline may expire before stream.setListener(jumpListener).
Expand All @@ -566,9 +587,63 @@ public void cancelled(Context context) {

context.addListener(new ServerStreamCancellationListener(), directExecutor());
}

private <ReqT, RespT> ServerCallParameters<ReqT, RespT> maySwitchExecutor(
final ServerMethodDefinition<ReqT, RespT> methodDef,
final ServerStream stream,
final Metadata headers,
final Context.CancellableContext context,
final Tag tag) {
final ServerCallImpl<ReqT, RespT> call = new ServerCallImpl<>(
stream,
methodDef.getMethodDescriptor(),
headers,
context,
decompressorRegistry,
compressorRegistry,
serverCallTracer,
tag);
final Executor switchingExecutor = executorSupplier.getExecutor(call, headers);
if (switchingExecutor != null && defaultExecutor instanceof SerializingExecutor) {
((SerializingExecutor)defaultExecutor).setExecutor(switchingExecutor);
}
return new ServerCallParameters<>(call, methodDef);
}
}

wrappedExecutor.execute(new StreamCreated());
final class CallHandled extends ContextRunnable {
CallHandled() {
super(context);
}

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

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

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

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

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.

"Fail to start server call");
stream.close(status, new Metadata());
context.cancel(null);
return;
}
listener = startWrappedCall(methodName, callParameters, headers);
} catch (Throwable ex) {
stream.close(Status.fromThrowable(ex), new Metadata());
context.cancel(null);
throw new IllegalStateException(ex);
} finally {
jumpListener.setListener(listener);
}
}
}

defaultExecutor.execute(new MethodLookup());
defaultExecutor.execute(new CallHandled());
}

private Context.CancellableContext createContext(
Expand All @@ -593,9 +668,8 @@ private Context.CancellableContext createContext(
}

/** Never returns {@code null}. */
private <ReqT, RespT> ServerStreamListener startCall(ServerStream stream, String fullMethodName,
ServerMethodDefinition<ReqT, RespT> methodDef, Metadata headers,
Context.CancellableContext context, StatsTraceContext statsTraceCtx, Tag tag) {
private <ReqT, RespT> ServerMethodDefinition<?,?> wrapMethod(ServerStream stream,
ServerMethodDefinition<ReqT, RespT> methodDef, StatsTraceContext statsTraceCtx) {
// TODO(ejona86): should we update fullMethodName to have the canonical path of the method?
statsTraceCtx.serverCallStarted(
new ServerCallInfoImpl<>(
Expand All @@ -609,34 +683,31 @@ private <ReqT, RespT> ServerStreamListener startCall(ServerStream stream, String
ServerMethodDefinition<ReqT, RespT> interceptedDef = methodDef.withServerCallHandler(handler);
ServerMethodDefinition<?, ?> wMethodDef = binlog == null
? interceptedDef : binlog.wrapMethodDefinition(interceptedDef);
return startWrappedCall(fullMethodName, wMethodDef, stream, headers, context, tag);
return wMethodDef;
}

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

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?


public ServerCallParameters(ServerCallImpl<ReqT, RespT> call,
ServerMethodDefinition<ReqT, RespT> methodDef) {
this.call = call;
this.methodDef = methodDef;
}
}

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

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

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.

ServerMethodDefinition<WReqT, WRespT> methodDef,
ServerStream stream,
Metadata headers,
Context.CancellableContext context,
Tag tag) {

ServerCallImpl<WReqT, WRespT> call = new ServerCallImpl<>(
stream,
methodDef.getMethodDescriptor(),
headers,
context,
decompressorRegistry,
compressorRegistry,
serverCallTracer,
tag);

ServerCall.Listener<WReqT> listener =
methodDef.getServerCallHandler().startCall(call, headers);
if (listener == null) {
ServerCallParameters<WReqT, WRespT> params,
Metadata headers) {
ServerCall.Listener<WReqT> callListener =
params.methodDef.getServerCallHandler().startCall(params.call, headers);
if (callListener == null) {
throw new NullPointerException(
"startCall() returned a null listener for method " + fullMethodName);
"startCall() returned a null listener for method " + fullMethodName);
}
return call.newServerStreamListener(listener);
return params.call.newServerStreamListener(callListener);
}
}

Expand Down