Skip to content

core: Delayed deserialization for unary/server-streaming calls #13004

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

Open
kannanjgithub wants to merge 3 commits into
base: master
Choose a base branch
from
+285 −16
Open

core: Delayed deserialization for unary/server-streaming calls #13004

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
09d8c90
Fix remote OOM via delayed deserialization for unary/server-streaming…
kannanjgithub Aug 25, 2026
553b33a
Address review comments.
kannanjgithub Aug 26, 2026
50197f1
Synchronously mark the call as cancelled by calling `closedInternal` …
kannanjgithub Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 60 additions & 6 deletions core/src/main/java/io/grpc/internal/ServerCallImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import io.grpc.CompressorRegistry;
import io.grpc.Context;
import io.grpc.DecompressorRegistry;
import io.grpc.Detachable;
import io.grpc.InternalDecompressorRegistry;
import io.grpc.InternalStatus;
import io.grpc.Metadata;
Expand All @@ -45,6 +46,9 @@
import io.perfmark.PerfMark;
import io.perfmark.Tag;
import io.perfmark.TaskCloseable;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.logging.Level;
import java.util.logging.Logger;
Expand Down Expand Up @@ -288,6 +292,7 @@ static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListene
private final ServerCallImpl<ReqT, ?> call;
private final ServerCall.Listener<ReqT> listener;
private final Context.CancellableContext context;
private InputStream delayedMessage;

public ServerStreamListenerImpl(
ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
Expand Down Expand Up @@ -320,6 +325,20 @@ public void messagesAvailable(MessageProducer producer) {
}
}

private static InputStream bufferMessage(InputStream is) throws IOException {
if (is instanceof Detachable) {
return ((Detachable) is).detach();
}
// Fallback: copy to byte array
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] buffer = new byte[4096];
int bytesRead;
while ((bytesRead = is.read(buffer)) != -1) {
baos.write(buffer, 0, bytesRead);
}
return new ByteArrayInputStream(baos.toByteArray());
}

@SuppressWarnings("Finally") // The code avoids suppressing the exception thrown from try
private void messagesAvailableInternal(final MessageProducer producer) {
if (call.cancelled) {
Expand All @@ -330,13 +349,31 @@ private void messagesAvailableInternal(final MessageProducer producer) {
InputStream message;
try {
while ((message = producer.next()) != null) {
try {
listener.onMessage(call.method.parseRequest(message));
} catch (Throwable t) {
GrpcUtil.closeQuietly(message);
throw t;
if (call.method.getType().clientSendsOneMessage()) {
if (delayedMessage != null) {
GrpcUtil.closeQuietly(message);
call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));
GrpcUtil.closeQuietly(delayedMessage);
delayedMessage = null;

@ejona86 ejona86 Aug 27, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This puts the call back into a normal state, so if other events happen after this one (e.g., message, or half close), that could end up propagating to the application before the cancel is processed. I don't know the easiest way to handle that though; obviously we could set some more state/booleans. It is probably worth looking into the exception handling in the executor see what would happen if we throw here.

@kannanjgithub kannanjgithub Aug 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

If an exception is thrown from messagesAvailableInternal, the catch block in the wrapped code submitted to the call executor catches it and calls internalClose(t) eventually leading to an asynchronous callback from the transport. It still does not handle the race you mentioned. Instead I'm now invoking closedInternal synchronously when the error is detected. This synchronously sets call.cancelled = true and cancels the context before returning from the executor task.

@ejona86 ejona86 Sep 5, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Now doesn't this code call the application's onCancel() twice? It calls it once here with the call to closedInternal(), and then later when the cancellation is actually processed.

closedInternal(Status.INTERNAL.withDescription("Too many requests"));
return;
}
try {
delayedMessage = bufferMessage(message);

@ejona86 ejona86 Aug 25, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Why are we making a copy here when we could just "not call close()" on the original message?

@kannanjgithub kannanjgithub Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

That would work for the Detachable InputStream since the ref count on Netty ByteBuf is still going to be non-zero even when detaching, so we might as well have just held the reference pass on to us.
But for non Detachable InputStream such as for compressed streams, if we don't copy to heap and release the message InputStream passed it will continue to hold the native memory for zlib objects which are allocated per request message and can be larger than the request message size itself.
By detaching the InputStream for Detachables, bufferMessage allows the close handling for both detachable and non-detachable cases be uniform without having to check which case it is.

@ejona86 ejona86 Sep 5, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

No need for detachable, and I would agree we shouldn't use it here. This is the code that calls close(), so we can just "not call close" if we want to avoid it. The compression context is relevant/important, but if we decompress then that in itself can hold a lot more memory than the compression context. We could make MessageDeframer return lazy inputstreams that call decompressor.decompress() when the first bytes are read.

It would be good to consider if forcing the transport to deal with this might be better, at least long-term. If MessageDeframer was told there was only one message, then it could delay running processBody() until closeWhenComplete == true (it will still run readRequiredBytes() until it got the message, though). It might be ugly, especially detecting cardinality violation (but dealing with it would be trivial: throw an exception), but probably worth checking at some point. (Binder mostly wouldn't even need any changes, because unary is already handled specially, but I don't remember if you can use streaming mode for a unary call.)

} catch (Throwable t) {
GrpcUtil.closeQuietly(message);
throw t;
}
message.close();
} else {
try {
listener.onMessage(call.method.parseRequest(message));
} catch (Throwable t) {
GrpcUtil.closeQuietly(message);
throw t;
}
message.close();
}
message.close();
}
} catch (Throwable t) {
GrpcUtil.closeQuietly(producer);
Expand All @@ -353,6 +390,19 @@ public void halfClosed() {
return;
}

if (delayedMessage != null) {
InputStream message = delayedMessage;
delayedMessage = null;
try {
listener.onMessage(call.method.parseRequest(message));
} catch (Throwable t) {
GrpcUtil.closeQuietly(message);
Throwables.throwIfUnchecked(t);
throw new RuntimeException(t);
}
GrpcUtil.closeQuietly(message);
}

listener.onHalfClose();
}
}
Expand All @@ -366,6 +416,10 @@ public void closed(Status status) {
}

private void closedInternal(Status status) {
if (delayedMessage != null) {
GrpcUtil.closeQuietly(delayedMessage);
delayedMessage = null;
}
Throwable cancelCause = null;
try {
if (status.isOk()) {
Expand Down
Loading
Loading