Skip to content

Commit df4db93

Browse files
kannanjgithubkannanjgithub
committed
Revert "core: Delayed deserialization for unary/server-streaming calls (grpc#13004) (grpc#13107)
This reverts commit 8ba4b48 because this change breaks grpc-kotlin tests that use InProcessTransport by causing a deadlock because [ServerCalls.kt](https://github.com/grpc/grpc-kotlin/blob/master/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt#L214-L215) only requests the next message after `onMessage` delivers the current one, and messages get queued because the client hasn't called `halfClose` yet. When the client halfCloses, `InProcessTransport.halfClose` [defers](https://github.com/grpc/grpc-java/blob/master/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java#L904) delivering halfClose if the server's receive queue is non-empty. **Resulting Deadlock:** - `ServerCallImpl` waits for `halfClosed()` before dispatching Message 1. - `InProcessTransport` waits for `call.request(1)` before dispatching `halfClosed()`. - The server waits for Message 1 before calling `call.request(1)`. - The test hangs indefinitely waiting for `closeStatus.await()` until the 10s test timeout expires. There are a few other failures too but those can be fixed in the test code alone. Internal Test failures: http://shortn/_If0NinnwQ4
1 parent 778f3a6 commit df4db93

4 files changed

Lines changed: 25 additions & 438 deletions

File tree

‎core/src/main/java/io/grpc/internal/MessageDeframer.java‎

Lines changed: 8 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -437,91 +437,14 @@ private InputStream getCompressedBody() {
437437
.asRuntimeException();
438438
}
439439

440-
return new LazyDecompressingInputStream(
441-
ReadableBuffers.openStream(nextFrame, true),
442-
maxInboundMessageSize,
443-
statsTraceCtx,
444-
decompressor);
445-
}
446-
447-
/**
448-
* An {@link InputStream} that delays decompressing a compressed frame until data is first read.
449-
*/
450-
@VisibleForTesting
451-
static final class LazyDecompressingInputStream extends FilterInputStream {
452-
private final Decompressor decompressor;
453-
private final int maxMessageSize;
454-
private final StatsTraceContext statsTraceCtx;
455-
private boolean initialized;
456-
private boolean closed;
457-
458-
LazyDecompressingInputStream(
459-
InputStream rawStream,
460-
int maxMessageSize,
461-
StatsTraceContext statsTraceCtx,
462-
Decompressor decompressor) {
463-
super(rawStream);
464-
this.decompressor = decompressor;
465-
this.maxMessageSize = maxMessageSize;
466-
this.statsTraceCtx = statsTraceCtx;
467-
}
468-
469-
private synchronized void ensureInitialized() throws IOException {
470-
if (closed) {
471-
throw new IOException("Stream closed");
472-
}
473-
if (!initialized) {
474-
InputStream decompressed = decompressor.decompress(in);
475-
in = new SizeEnforcingInputStream(decompressed, maxMessageSize, statsTraceCtx);
476-
initialized = true;
477-
}
478-
}
479-
480-
@Override
481-
public int read() throws IOException {
482-
ensureInitialized();
483-
return super.read();
484-
}
485-
486-
@Override
487-
public int read(byte[] b, int off, int len) throws IOException {
488-
ensureInitialized();
489-
return super.read(b, off, len);
490-
}
491-
492-
@Override
493-
public long skip(long n) throws IOException {
494-
ensureInitialized();
495-
return super.skip(n);
496-
}
497-
498-
@Override
499-
public int available() throws IOException {
500-
ensureInitialized();
501-
return super.available();
502-
}
503-
504-
@Override
505-
public synchronized void close() throws IOException {
506-
if (!closed) {
507-
closed = true;
508-
super.close();
509-
}
510-
}
511-
512-
@Override
513-
public synchronized void mark(int readlimit) {
514-
// Mark is not supported
515-
}
516-
517-
@Override
518-
public synchronized void reset() throws IOException {
519-
throw new IOException("Mark not supported");
520-
}
521-
522-
@Override
523-
public boolean markSupported() {
524-
return false;
440+
try {
441+
// Enforce the maxMessageSize limit on the returned stream.
442+
InputStream unlimitedStream =
443+
decompressor.decompress(ReadableBuffers.openStream(nextFrame, true));
444+
return new SizeEnforcingInputStream(
445+
unlimitedStream, maxInboundMessageSize, statsTraceCtx);
446+
} catch (IOException e) {
447+
throw new RuntimeException(e);
525448
}
526449
}
527450

‎core/src/main/java/io/grpc/internal/ServerCallImpl.java‎

Lines changed: 6 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -288,7 +288,6 @@ static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListene
288288
private final ServerCallImpl<ReqT, ?> call;
289289
private final ServerCall.Listener<ReqT> listener;
290290
private final Context.CancellableContext context;
291-
private InputStream delayedMessage;
292291

293292
public ServerStreamListenerImpl(
294293
ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
@@ -331,27 +330,13 @@ private void messagesAvailableInternal(final MessageProducer producer) {
331330
InputStream message;
332331
try {
333332
while ((message = producer.next()) != null) {
334-
// TODO: Consider forcing this check to be done in the transport (MessageDeframer)
335-
// https://github.com/grpc/grpc-java/pull/13004/changes#r3939373996
336-
if (call.method.getType().clientSendsOneMessage()) {
337-
if (delayedMessage != null) {
338-
GrpcUtil.closeQuietly(message);
339-
call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));
340-
GrpcUtil.closeQuietly(delayedMessage);
341-
delayedMessage = null;
342-
call.cancelled = true;
343-
return;
344-
}
345-
delayedMessage = message;
346-
} else {
347-
try {
348-
listener.onMessage(call.method.parseRequest(message));
349-
} catch (Throwable t) {
350-
GrpcUtil.closeQuietly(message);
351-
throw t;
352-
}
353-
message.close();
333+
try {
334+
listener.onMessage(call.method.parseRequest(message));
335+
} catch (Throwable t) {
336+
GrpcUtil.closeQuietly(message);
337+
throw t;
354338
}
339+
message.close();
355340
}
356341
} catch (Throwable t) {
357342
GrpcUtil.closeQuietly(producer);
@@ -368,19 +353,6 @@ public void halfClosed() {
368353
return;
369354
}
370355

371-
if (delayedMessage != null) {
372-
InputStream message = delayedMessage;
373-
delayedMessage = null;
374-
try {
375-
listener.onMessage(call.method.parseRequest(message));
376-
} catch (Throwable t) {
377-
GrpcUtil.closeQuietly(message);
378-
Throwables.throwIfUnchecked(t);
379-
throw new RuntimeException(t);
380-
}
381-
GrpcUtil.closeQuietly(message);
382-
}
383-
384356
listener.onHalfClose();
385357
}
386358
}
@@ -394,10 +366,6 @@ public void closed(Status status) {
394366
}
395367

396368
private void closedInternal(Status status) {
397-
if (delayedMessage != null) {
398-
GrpcUtil.closeQuietly(delayedMessage);
399-
delayedMessage = null;
400-
}
401369
Throwable cancelCause = null;
402370
try {
403371
if (status.isOk()) {

‎core/src/test/java/io/grpc/internal/MessageDeframerTest.java‎

Lines changed: 0 additions & 150 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,6 @@
1919
import static com.google.common.truth.Truth.assertThat;
2020
import static io.grpc.internal.GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE;
2121
import static org.junit.Assert.assertEquals;
22-
import static org.junit.Assert.assertFalse;
23-
import static org.junit.Assert.assertNotNull;
2422
import static org.junit.Assert.assertNull;
2523
import static org.junit.Assert.assertThrows;
2624
import static org.junit.Assert.assertTrue;
@@ -37,11 +35,9 @@
3735
import com.google.common.io.ByteStreams;
3836
import com.google.common.primitives.Bytes;
3937
import io.grpc.Codec;
40-
import io.grpc.Decompressor;
4138
import io.grpc.InternalChannelz.TransportStats;
4239
import io.grpc.StatusRuntimeException;
4340
import io.grpc.StreamTracer;
44-
import io.grpc.internal.MessageDeframer.LazyDecompressingInputStream;
4541
import io.grpc.internal.MessageDeframer.Listener;
4642
import io.grpc.internal.MessageDeframer.SizeEnforcingInputStream;
4743
import io.grpc.internal.testing.TestStreamTracer.TestBaseStreamTracer;
@@ -56,7 +52,6 @@
5652
import java.util.List;
5753
import java.util.Locale;
5854
import java.util.concurrent.TimeUnit;
59-
import java.util.concurrent.atomic.AtomicBoolean;
6055
import java.util.zip.GZIPOutputStream;
6156
import org.junit.Before;
6257
import org.junit.Test;
@@ -318,75 +313,6 @@ public void compressed() {
318313
verifyNoMoreInteractions(listener);
319314
}
320315

321-
@Test
322-
public void compressed_lazyDecompression() throws IOException {
323-
final AtomicBoolean decompressCalled = new AtomicBoolean(false);
324-
Decompressor countingDecompressor = new Decompressor() {
325-
@Override
326-
public String getMessageEncoding() {
327-
return "gzip";
328-
}
329-
330-
@Override
331-
public InputStream decompress(InputStream is) throws IOException {
332-
decompressCalled.set(true);
333-
return new Codec.Gzip().decompress(is);
334-
}
335-
};
336-
337-
deframer = new MessageDeframer(listener, countingDecompressor, DEFAULT_MAX_MESSAGE_SIZE,
338-
statsTraceCtx, transportTracer);
339-
deframer.request(1);
340-
341-
byte[] payload = compress(new byte[1000]);
342-
byte[] header = new byte[]{1, 0, 0, 0, (byte) payload.length};
343-
deframer.deframe(buffer(Bytes.concat(header, payload)));
344-
345-
verify(listener).messagesAvailable(producer.capture());
346-
InputStream stream = producer.getValue().next();
347-
assertNotNull(stream);
348-
349-
// Decompressor should not be invoked before bytes are read
350-
assertFalse(decompressCalled.get());
351-
352-
// Reading a byte triggers decompression
353-
assertEquals(0, stream.read());
354-
assertTrue(decompressCalled.get());
355-
}
356-
357-
@Test
358-
public void compressed_closeWithoutReading_noDecompression() throws IOException {
359-
final AtomicBoolean decompressCalled = new AtomicBoolean(false);
360-
Decompressor countingDecompressor = new Decompressor() {
361-
@Override
362-
public String getMessageEncoding() {
363-
return "gzip";
364-
}
365-
366-
@Override
367-
public InputStream decompress(InputStream is) throws IOException {
368-
decompressCalled.set(true);
369-
return new Codec.Gzip().decompress(is);
370-
}
371-
};
372-
373-
deframer = new MessageDeframer(listener, countingDecompressor, DEFAULT_MAX_MESSAGE_SIZE,
374-
statsTraceCtx, transportTracer);
375-
deframer.request(1);
376-
377-
byte[] payload = compress(new byte[1000]);
378-
byte[] header = new byte[]{1, 0, 0, 0, (byte) payload.length};
379-
deframer.deframe(buffer(Bytes.concat(header, payload)));
380-
381-
verify(listener).messagesAvailable(producer.capture());
382-
InputStream stream = producer.getValue().next();
383-
assertNotNull(stream);
384-
385-
// Closing without reading should not decompress
386-
stream.close();
387-
assertFalse(decompressCalled.get());
388-
}
389-
390316
@Test
391317
public void deliverIsReentrantSafe() {
392318
doAnswer(
@@ -567,82 +493,6 @@ public void sizeEnforcingInputStream_markReset() throws IOException {
567493
}
568494
}
569495

570-
@RunWith(JUnit4.class)
571-
public static class LazyDecompressingInputStreamTests {
572-
private TestBaseStreamTracer tracer = new TestBaseStreamTracer();
573-
private StatsTraceContext statsTraceCtx = new StatsTraceContext(new StreamTracer[]{tracer});
574-
575-
@Test
576-
public void lazyDecompressingInputStream_doesNotInitializeUntilRead() throws IOException {
577-
final AtomicBoolean decompressCalled = new AtomicBoolean(false);
578-
Decompressor countingDecompressor = new Decompressor() {
579-
@Override
580-
public String getMessageEncoding() {
581-
return "gzip";
582-
}
583-
584-
@Override
585-
public InputStream decompress(InputStream is) throws IOException {
586-
decompressCalled.set(true);
587-
return new Codec.Gzip().decompress(is);
588-
}
589-
};
590-
591-
ByteArrayInputStream in =
592-
new ByteArrayInputStream(compress("hello".getBytes(StandardCharsets.UTF_8)));
593-
LazyDecompressingInputStream stream = new LazyDecompressingInputStream(
594-
in, 100, statsTraceCtx, countingDecompressor);
595-
596-
assertFalse(stream.markSupported());
597-
stream.mark(10);
598-
assertThrows(IOException.class, () -> stream.reset());
599-
assertFalse(decompressCalled.get());
600-
byte[] buf = new byte[5];
601-
int read = stream.read(buf);
602-
assertEquals(5, read);
603-
assertEquals("hello", new String(buf, StandardCharsets.UTF_8));
604-
assertTrue(decompressCalled.get());
605-
stream.close();
606-
}
607-
608-
@Test
609-
public void lazyDecompressingInputStream_closeWithoutRead() throws IOException {
610-
final AtomicBoolean decompressCalled = new AtomicBoolean(false);
611-
final AtomicBoolean inClosed = new AtomicBoolean(false);
612-
Decompressor countingDecompressor = new Decompressor() {
613-
@Override
614-
public String getMessageEncoding() {
615-
return "gzip";
616-
}
617-
618-
@Override
619-
public InputStream decompress(InputStream is) throws IOException {
620-
decompressCalled.set(true);
621-
return new Codec.Gzip().decompress(is);
622-
}
623-
};
624-
625-
ByteArrayInputStream in =
626-
new ByteArrayInputStream(compress("hello".getBytes(StandardCharsets.UTF_8))) {
627-
@Override
628-
public void close() throws IOException {
629-
inClosed.set(true);
630-
super.close();
631-
}
632-
};
633-
LazyDecompressingInputStream stream = new LazyDecompressingInputStream(
634-
in, 100, statsTraceCtx, countingDecompressor);
635-
636-
assertFalse(decompressCalled.get());
637-
stream.close();
638-
assertTrue(inClosed.get());
639-
assertFalse(decompressCalled.get());
640-
641-
// Reading after close should throw IOException
642-
assertThrows(IOException.class, () -> stream.read());
643-
}
644-
}
645-
646496
/**
647497
* Verify stats were published through the tracer.
648498
*

0 commit comments

Comments
 (0)