From 1b3569167dcd9566a21b083209c16f0f7401cb9a Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 9 Oct 2026 11:22:08 +0000 Subject: [PATCH] Revert "core: Delayed deserialization for unary/server-streaming calls (#13004)" This reverts commit 8ba4b487217703ba60b77c080baabda58c790fc8. --- .../io/grpc/internal/MessageDeframer.java | 93 +-------- .../java/io/grpc/internal/ServerCallImpl.java | 44 +---- .../io/grpc/internal/MessageDeframerTest.java | 150 --------------- .../io/grpc/internal/ServerCallImplTest.java | 176 ++---------------- 4 files changed, 25 insertions(+), 438 deletions(-) diff --git a/core/src/main/java/io/grpc/internal/MessageDeframer.java b/core/src/main/java/io/grpc/internal/MessageDeframer.java index 3f2c408cc9e..f388c006e97 100644 --- a/core/src/main/java/io/grpc/internal/MessageDeframer.java +++ b/core/src/main/java/io/grpc/internal/MessageDeframer.java @@ -437,91 +437,14 @@ private InputStream getCompressedBody() { .asRuntimeException(); } - return new LazyDecompressingInputStream( - ReadableBuffers.openStream(nextFrame, true), - maxInboundMessageSize, - statsTraceCtx, - decompressor); - } - - /** - * An {@link InputStream} that delays decompressing a compressed frame until data is first read. - */ - @VisibleForTesting - static final class LazyDecompressingInputStream extends FilterInputStream { - private final Decompressor decompressor; - private final int maxMessageSize; - private final StatsTraceContext statsTraceCtx; - private boolean initialized; - private boolean closed; - - LazyDecompressingInputStream( - InputStream rawStream, - int maxMessageSize, - StatsTraceContext statsTraceCtx, - Decompressor decompressor) { - super(rawStream); - this.decompressor = decompressor; - this.maxMessageSize = maxMessageSize; - this.statsTraceCtx = statsTraceCtx; - } - - private synchronized void ensureInitialized() throws IOException { - if (closed) { - throw new IOException("Stream closed"); - } - if (!initialized) { - InputStream decompressed = decompressor.decompress(in); - in = new SizeEnforcingInputStream(decompressed, maxMessageSize, statsTraceCtx); - initialized = true; - } - } - - @Override - public int read() throws IOException { - ensureInitialized(); - return super.read(); - } - - @Override - public int read(byte[] b, int off, int len) throws IOException { - ensureInitialized(); - return super.read(b, off, len); - } - - @Override - public long skip(long n) throws IOException { - ensureInitialized(); - return super.skip(n); - } - - @Override - public int available() throws IOException { - ensureInitialized(); - return super.available(); - } - - @Override - public synchronized void close() throws IOException { - if (!closed) { - closed = true; - super.close(); - } - } - - @Override - public synchronized void mark(int readlimit) { - // Mark is not supported - } - - @Override - public synchronized void reset() throws IOException { - throw new IOException("Mark not supported"); - } - - @Override - public boolean markSupported() { - return false; + try { + // Enforce the maxMessageSize limit on the returned stream. + InputStream unlimitedStream = + decompressor.decompress(ReadableBuffers.openStream(nextFrame, true)); + return new SizeEnforcingInputStream( + unlimitedStream, maxInboundMessageSize, statsTraceCtx); + } catch (IOException e) { + throw new RuntimeException(e); } } diff --git a/core/src/main/java/io/grpc/internal/ServerCallImpl.java b/core/src/main/java/io/grpc/internal/ServerCallImpl.java index f15ad1d4f1f..1f321d0d8dc 100644 --- a/core/src/main/java/io/grpc/internal/ServerCallImpl.java +++ b/core/src/main/java/io/grpc/internal/ServerCallImpl.java @@ -296,7 +296,6 @@ static final class ServerStreamListenerImpl implements ServerStreamListene private final ServerCallImpl call; private final ServerCall.Listener listener; private final Context.CancellableContext context; - private InputStream delayedMessage; public ServerStreamListenerImpl( ServerCallImpl call, ServerCall.Listener listener, @@ -339,27 +338,13 @@ private void messagesAvailableInternal(final MessageProducer producer) { InputStream message; try { while ((message = producer.next()) != null) { - // TODO: Consider forcing this check to be done in the transport (MessageDeframer) - // https://github.com/grpc/grpc-java/pull/13004/changes#r3939373996 - 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; - call.cancelled = true; - return; - } - delayedMessage = message; - } else { - try { - listener.onMessage(call.method.parseRequest(message)); - } catch (Throwable t) { - GrpcUtil.closeQuietly(message); - throw t; - } - message.close(); + try { + listener.onMessage(call.method.parseRequest(message)); + } catch (Throwable t) { + GrpcUtil.closeQuietly(message); + throw t; } + message.close(); } } catch (Throwable t) { GrpcUtil.closeQuietly(producer); @@ -376,19 +361,6 @@ 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(); } } @@ -402,10 +374,6 @@ 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()) { diff --git a/core/src/test/java/io/grpc/internal/MessageDeframerTest.java b/core/src/test/java/io/grpc/internal/MessageDeframerTest.java index b279f90301a..54758bc096f 100644 --- a/core/src/test/java/io/grpc/internal/MessageDeframerTest.java +++ b/core/src/test/java/io/grpc/internal/MessageDeframerTest.java @@ -19,8 +19,6 @@ import static com.google.common.truth.Truth.assertThat; import static io.grpc.internal.GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -37,11 +35,9 @@ import com.google.common.io.ByteStreams; import com.google.common.primitives.Bytes; import io.grpc.Codec; -import io.grpc.Decompressor; import io.grpc.InternalChannelz.TransportStats; import io.grpc.StatusRuntimeException; import io.grpc.StreamTracer; -import io.grpc.internal.MessageDeframer.LazyDecompressingInputStream; import io.grpc.internal.MessageDeframer.Listener; import io.grpc.internal.MessageDeframer.SizeEnforcingInputStream; import io.grpc.internal.testing.TestStreamTracer.TestBaseStreamTracer; @@ -56,7 +52,6 @@ import java.util.List; import java.util.Locale; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.zip.GZIPOutputStream; import org.junit.Before; import org.junit.Test; @@ -318,75 +313,6 @@ public void compressed() { verifyNoMoreInteractions(listener); } - @Test - public void compressed_lazyDecompression() throws IOException { - final AtomicBoolean decompressCalled = new AtomicBoolean(false); - Decompressor countingDecompressor = new Decompressor() { - @Override - public String getMessageEncoding() { - return "gzip"; - } - - @Override - public InputStream decompress(InputStream is) throws IOException { - decompressCalled.set(true); - return new Codec.Gzip().decompress(is); - } - }; - - deframer = new MessageDeframer(listener, countingDecompressor, DEFAULT_MAX_MESSAGE_SIZE, - statsTraceCtx, transportTracer); - deframer.request(1); - - byte[] payload = compress(new byte[1000]); - byte[] header = new byte[]{1, 0, 0, 0, (byte) payload.length}; - deframer.deframe(buffer(Bytes.concat(header, payload))); - - verify(listener).messagesAvailable(producer.capture()); - InputStream stream = producer.getValue().next(); - assertNotNull(stream); - - // Decompressor should not be invoked before bytes are read - assertFalse(decompressCalled.get()); - - // Reading a byte triggers decompression - assertEquals(0, stream.read()); - assertTrue(decompressCalled.get()); - } - - @Test - public void compressed_closeWithoutReading_noDecompression() throws IOException { - final AtomicBoolean decompressCalled = new AtomicBoolean(false); - Decompressor countingDecompressor = new Decompressor() { - @Override - public String getMessageEncoding() { - return "gzip"; - } - - @Override - public InputStream decompress(InputStream is) throws IOException { - decompressCalled.set(true); - return new Codec.Gzip().decompress(is); - } - }; - - deframer = new MessageDeframer(listener, countingDecompressor, DEFAULT_MAX_MESSAGE_SIZE, - statsTraceCtx, transportTracer); - deframer.request(1); - - byte[] payload = compress(new byte[1000]); - byte[] header = new byte[]{1, 0, 0, 0, (byte) payload.length}; - deframer.deframe(buffer(Bytes.concat(header, payload))); - - verify(listener).messagesAvailable(producer.capture()); - InputStream stream = producer.getValue().next(); - assertNotNull(stream); - - // Closing without reading should not decompress - stream.close(); - assertFalse(decompressCalled.get()); - } - @Test public void deliverIsReentrantSafe() { doAnswer( @@ -567,82 +493,6 @@ public void sizeEnforcingInputStream_markReset() throws IOException { } } - @RunWith(JUnit4.class) - public static class LazyDecompressingInputStreamTests { - private TestBaseStreamTracer tracer = new TestBaseStreamTracer(); - private StatsTraceContext statsTraceCtx = new StatsTraceContext(new StreamTracer[]{tracer}); - - @Test - public void lazyDecompressingInputStream_doesNotInitializeUntilRead() throws IOException { - final AtomicBoolean decompressCalled = new AtomicBoolean(false); - Decompressor countingDecompressor = new Decompressor() { - @Override - public String getMessageEncoding() { - return "gzip"; - } - - @Override - public InputStream decompress(InputStream is) throws IOException { - decompressCalled.set(true); - return new Codec.Gzip().decompress(is); - } - }; - - ByteArrayInputStream in = - new ByteArrayInputStream(compress("hello".getBytes(StandardCharsets.UTF_8))); - LazyDecompressingInputStream stream = new LazyDecompressingInputStream( - in, 100, statsTraceCtx, countingDecompressor); - - assertFalse(stream.markSupported()); - stream.mark(10); - assertThrows(IOException.class, () -> stream.reset()); - assertFalse(decompressCalled.get()); - byte[] buf = new byte[5]; - int read = stream.read(buf); - assertEquals(5, read); - assertEquals("hello", new String(buf, StandardCharsets.UTF_8)); - assertTrue(decompressCalled.get()); - stream.close(); - } - - @Test - public void lazyDecompressingInputStream_closeWithoutRead() throws IOException { - final AtomicBoolean decompressCalled = new AtomicBoolean(false); - final AtomicBoolean inClosed = new AtomicBoolean(false); - Decompressor countingDecompressor = new Decompressor() { - @Override - public String getMessageEncoding() { - return "gzip"; - } - - @Override - public InputStream decompress(InputStream is) throws IOException { - decompressCalled.set(true); - return new Codec.Gzip().decompress(is); - } - }; - - ByteArrayInputStream in = - new ByteArrayInputStream(compress("hello".getBytes(StandardCharsets.UTF_8))) { - @Override - public void close() throws IOException { - inClosed.set(true); - super.close(); - } - }; - LazyDecompressingInputStream stream = new LazyDecompressingInputStream( - in, 100, statsTraceCtx, countingDecompressor); - - assertFalse(decompressCalled.get()); - stream.close(); - assertTrue(inClosed.get()); - assertFalse(decompressCalled.get()); - - // Reading after close should throw IOException - assertThrows(IOException.class, () -> stream.read()); - } - } - /** * Verify stats were published through the tracer. * diff --git a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java index babf8db8f2c..b29b6a1f6b7 100644 --- a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java +++ b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java @@ -51,7 +51,6 @@ import io.grpc.internal.ServerCallImpl.ServerStreamListenerImpl; import io.perfmark.PerfMark; import java.io.ByteArrayInputStream; -import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import org.junit.Before; @@ -85,23 +84,7 @@ public class ServerCallImplTest { private static final MethodDescriptor CLIENT_STREAMING_METHOD = MethodDescriptor.newBuilder() - .setType(MethodType.CLIENT_STREAMING) - .setFullMethodName("service/method") - .setRequestMarshaller(new LongMarshaller()) - .setResponseMarshaller(new LongMarshaller()) - .build(); - - private static final MethodDescriptor BIDI_STREAMING_METHOD = - MethodDescriptor.newBuilder() - .setType(MethodType.BIDI_STREAMING) - .setFullMethodName("service/method") - .setRequestMarshaller(new LongMarshaller()) - .setResponseMarshaller(new LongMarshaller()) - .build(); - - private static final MethodDescriptor SERVER_STREAMING_METHOD = - MethodDescriptor.newBuilder() - .setType(MethodType.SERVER_STREAMING) + .setType(MethodType.UNARY) .setFullMethodName("service/method") .setRequestMarshaller(new LongMarshaller()) .setResponseMarshaller(new LongMarshaller()) @@ -473,153 +456,41 @@ public void streamListener_onReady_onlyOnce() { } @Test - public void streamListener_messageRead_unary_delayed() { + public void streamListener_messageRead() { ServerStreamListenerImpl streamListener = new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); - FakeCloseTrackerInputStream messageStream = - new FakeCloseTrackerInputStream(UNARY_METHOD.streamRequest(1234L), false); - streamListener.messagesAvailable(new SingleMessageProducer(messageStream)); - - // Message should not be delivered or closed yet - verify(callListener, never()).onMessage(any(Long.class)); - assertFalse(messageStream.closed); - - streamListener.halfClosed(); - - // Now it should be delivered and closed - verify(callListener).onMessage(1234L); - verify(callListener).onHalfClose(); - assertTrue(messageStream.closed); - } - - @Test - public void streamListener_messageRead_serverStreaming_delayed() { - call = new ServerCallImpl<>(stream, SERVER_STREAMING_METHOD, requestHeaders, context, - DecompressorRegistry.getDefaultInstance(), CompressorRegistry.getDefaultInstance(), - serverCallTracer, PerfMark.createTag()); - ServerStreamListenerImpl streamListener = - new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); - FakeCloseTrackerInputStream messageStream = - new FakeCloseTrackerInputStream( - SERVER_STREAMING_METHOD.streamRequest(1234L), false); - streamListener.messagesAvailable(new SingleMessageProducer(messageStream)); - - // Message should not be delivered or closed yet - verify(callListener, never()).onMessage(any(Long.class)); - assertFalse(messageStream.closed); - - streamListener.halfClosed(); - - // Now it should be delivered and closed - verify(callListener).onMessage(1234L); - verify(callListener).onHalfClose(); - assertTrue(messageStream.closed); - } - - @Test - public void streamListener_messageRead_bidi_notDelayed() { - call = new ServerCallImpl<>(stream, BIDI_STREAMING_METHOD, requestHeaders, context, - DecompressorRegistry.getDefaultInstance(), CompressorRegistry.getDefaultInstance(), - serverCallTracer, PerfMark.createTag()); - ServerStreamListenerImpl streamListener = - new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); - streamListener.messagesAvailable( - new SingleMessageProducer(BIDI_STREAMING_METHOD.streamRequest(1234L))); + streamListener.messagesAvailable(new SingleMessageProducer(UNARY_METHOD.streamRequest(1234L))); - // Message should be delivered immediately verify(callListener).onMessage(1234L); - verify(callListener, never()).onHalfClose(); } @Test - public void streamListener_messageRead_unary_tooManyRequests() { + public void streamListener_messageRead_onlyOnce() { ServerStreamListenerImpl streamListener = new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); streamListener.messagesAvailable(new SingleMessageProducer(UNARY_METHOD.streamRequest(1234L))); - - // Sending second message should fail - streamListener.messagesAvailable(new SingleMessageProducer(UNARY_METHOD.streamRequest(5678L))); - - verify(stream).cancel(any(Status.class)); - verify(callListener, never()).onMessage(any(Long.class)); - assertTrue(call.isCancelled()); - - // Subsequent halfClosed should be ignored because call is cancelled - streamListener.halfClosed(); - verify(callListener, never()).onHalfClose(); - - // When transport closes, onCancel is called once - streamListener.closed(Status.CANCELLED); - verify(callListener).onCancel(); - assertTrue(context.isCancelled()); - } - - @Test - public void streamListener_messageRead_onlyOnce_unary() { - ServerStreamListenerImpl streamListener = - new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); - FakeCloseTrackerInputStream messageStream = - new FakeCloseTrackerInputStream(UNARY_METHOD.streamRequest(1234L), false); - streamListener.messagesAvailable(new SingleMessageProducer(messageStream)); - - assertFalse(messageStream.closed); - - // canceling the call should clean up and prevent delivery + // canceling the call should short circuit future halfClosed() calls. streamListener.closed(Status.CANCELLED); - streamListener.halfClosed(); + streamListener.messagesAvailable(new SingleMessageProducer(UNARY_METHOD.streamRequest(1234L))); - verify(callListener, never()).onMessage(any(Long.class)); - assertTrue(messageStream.closed); + verify(callListener).onMessage(1234L); } @Test - public void streamListener_unexpectedRuntimeException_unary() { + public void streamListener_unexpectedRuntimeException() { ServerStreamListenerImpl streamListener = new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); doThrow(new RuntimeException("unexpected exception")) .when(callListener) .onMessage(any(Long.class)); - FakeCloseTrackerInputStream messageStream = - new FakeCloseTrackerInputStream(UNARY_METHOD.streamRequest(1234L), false); - - streamListener.messagesAvailable(new SingleMessageProducer(messageStream)); + InputStream inputStream = UNARY_METHOD.streamRequest(1234L); - // Exception should not be thrown yet because deserialization/delivery is delayed - verify(callListener, never()).onMessage(any(Long.class)); - assertFalse(messageStream.closed); - - // It should be thrown during halfClosed + SingleMessageProducer producer = new SingleMessageProducer(inputStream); RuntimeException e = assertThrows(RuntimeException.class, - () -> streamListener.halfClosed()); + () -> streamListener.messagesAvailable(producer)); assertThat(e).hasMessageThat().isEqualTo("unexpected exception"); - - // The stream should have been closed in the catch block - assertTrue(messageStream.closed); - } - - @Test - public void streamListener_halfClosed_closeException() { - ServerStreamListenerImpl streamListener = - new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); - - FakeCloseTrackerInputStream messageStream = - new FakeCloseTrackerInputStream(UNARY_METHOD.streamRequest(1234L), true); - - streamListener.messagesAvailable(new SingleMessageProducer(messageStream)); - - // Message should not be delivered yet - verify(callListener, never()).onMessage(any(Long.class)); - assertFalse(messageStream.closed); - - // halfClosed should not throw because we use closeQuietly - streamListener.halfClosed(); - - // The message was delivered and halfClosed completed - verify(callListener).onMessage(1234L); - verify(callListener).onHalfClose(); - assertTrue(messageStream.closed); } @Test @@ -672,29 +543,4 @@ public Long parse(InputStream stream) { } } } - - private static class FakeCloseTrackerInputStream extends InputStream { - boolean closed = false; - private final InputStream delegate; - private final boolean throwOnClose; - - FakeCloseTrackerInputStream(InputStream delegate, boolean throwOnClose) { - this.delegate = delegate; - this.throwOnClose = throwOnClose; - } - - @Override - public int read() throws IOException { - return delegate.read(); - } - - @Override - public void close() throws IOException { - closed = true; - if (throwOnClose) { - throw new IOException("close failed"); - } - delegate.close(); - } - } }