Skip to content

netty: Ignore inbound frames without a transport state #13073

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
NianJiuZst wants to merge 1 commit into
base: master
Choose a base branch
from
+102 −11
Open

netty: Ignore inbound frames without a transport state #13073

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
26 changes: 16 additions & 10 deletions netty/src/main/java/io/grpc/netty/NettyClientHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -435,9 +435,15 @@ void returnProcessedBytes(Http2Stream stream, int bytes) {
}

private void onHeadersRead(int streamId, Http2Headers headers, boolean endStream) {
if (keepAliveManager != null) {
keepAliveManager.onDataReceived();
}
// Stream 1 is reserved for the Upgrade response, so we should ignore its headers here:
if (streamId != Http2CodecUtil.HTTP_UPGRADE_STREAM_ID) {
NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId));
if (stream == null) {
return;
}
PerfMark.event("NettyClientHandler.onHeadersRead", stream.tag());
// check metadata size vs soft limit
int h2HeadersSize = Utils.getH2HeadersSize(headers);
Expand All @@ -463,23 +469,24 @@ private void onHeadersRead(int streamId, Http2Headers headers, boolean endStream
}
stream.transportHeadersReceived(headers, endStream);
}

if (keepAliveManager != null) {
keepAliveManager.onDataReceived();
}
}

/**
* Handler for an inbound HTTP/2 DATA frame.
*/
private void onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream) {
private int onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream) {
flowControlPing().onDataRead(data.readableBytes(), padding);
NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId));
PerfMark.event("NettyClientHandler.onDataRead", stream.tag());
stream.transportDataReceived(data, endOfStream);
if (keepAliveManager != null) {
keepAliveManager.onDataReceived();
}
NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId));
if (stream == null) {
// No deframer will consume this payload, so return its flow-control credit immediately.
return data.readableBytes() + padding;
}
PerfMark.event("NettyClientHandler.onDataRead", stream.tag());
stream.transportDataReceived(data, endOfStream);
return padding;
}

/**
Expand Down Expand Up @@ -1089,8 +1096,7 @@ public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
@Override
public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
boolean endOfStream) throws Http2Exception {
NettyClientHandler.this.onDataRead(streamId, data, padding, endOfStream);
return padding;
return NettyClientHandler.this.onDataRead(streamId, data, padding, endOfStream);
}

@Override
Expand Down
87 changes: 86 additions & 1 deletion netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.handler.codec.http2.DefaultHttp2Connection;
import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.Http2Connection;
import io.netty.handler.codec.http2.Http2Error;
Expand Down Expand Up @@ -135,7 +136,10 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase<NettyClientHand
private KeepAliveManager mockKeepAliveManager = null;
private List<String> setKeepaliveManagerFor = ImmutableList.of("cancelShouldSucceed",
"sendFrameShouldSucceed", "channelShutdownShouldCancelBufferedStreams",
"createIncrementsIdsForActualAndBufferdStreams", "dataPingAckIsRecognized");
"createIncrementsIdsForActualAndBufferdStreams", "dataPingAckIsRecognized",
"headersWithoutTransportStateShouldBeIgnored", "trailersWithoutTransportStateShouldBeIgnored",
"dataWithoutTransportStateShouldBeConsumed",
"dataEndStreamWithoutTransportStateShouldBeConsumed");
private Runnable tooManyPingsRunnable = new Runnable() {
@Override public void run() {}
};
Expand Down Expand Up @@ -372,6 +376,87 @@ public void sendForUnknownStreamShouldFail() throws Exception {
assertFalse(future.isSuccess());
}

@Test
public void headersWithoutTransportStateShouldBeIgnored() throws Exception {
headersWithoutTransportState(false);
}

@Test
public void trailersWithoutTransportStateShouldBeIgnored() throws Exception {
headersWithoutTransportState(true);
}

private void headersWithoutTransportState(boolean endStream) throws Exception {
createStream();
int streamWithoutTransportState = STREAM_ID + 2;
Http2Stream http2Stream = connection().local().createStream(streamWithoutTransportState, true);
Http2Headers headers = new DefaultHttp2Headers();
if (endStream) {
http2Stream.headersReceived(false);
headers.set("grpc-status", "0");
} else {
headers.status(STATUS_OK).set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC);
}
ChannelHandlerContext frameContext = newMockContext();
new DefaultHttp2FrameWriter().writeHeaders(
frameContext, streamWithoutTransportState, headers, 0, endStream, newPromise());

ByteBuf frame = captureWrite(frameContext);
channelRead(frame);

assertEquals(0, frame.refCnt());
assertNull(lifecycleManager.getShutdownStatus());
assertTrue(channel().isOpen());
verify(mockKeepAliveManager).onDataReceived();
verify(streamListener, never()).closed(any(Status.class), any(RpcProgress.class),
any(Metadata.class));
verify(streamListener, never()).headersRead(any(Metadata.class));

channelRead(headersFrame(STREAM_ID, new DefaultHttp2Headers().status(STATUS_OK)
.set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)));
verify(streamListener).headersRead(any(Metadata.class));
}

@Test
public void dataWithoutTransportStateShouldBeConsumed() throws Exception {
dataWithoutTransportState(false);
}

@Test
public void dataEndStreamWithoutTransportStateShouldBeConsumed() throws Exception {
dataWithoutTransportState(true);
}

private void dataWithoutTransportState(boolean endStream) throws Exception {
createStream();
int streamWithoutTransportState = STREAM_ID + 2;
Http2Stream http2Stream = connection().local().createStream(streamWithoutTransportState, true);
http2Stream.headersReceived(false);
Http2LocalFlowController flowController = connection().local().flowController();
ChannelHandlerContext frameContext = newMockContext();
ByteBuf data = content();
new DefaultHttp2FrameWriter().writeData(
frameContext, streamWithoutTransportState, data, 5, endStream, newPromise());

ByteBuf frame = captureWrite(frameContext);
channelRead(frame);

assertEquals(0, frame.refCnt());
assertEquals(0, data.refCnt());
assertNull(lifecycleManager.getShutdownStatus());
assertTrue(channel().isOpen());
assertEquals(0, flowController.unconsumedBytes(http2Stream));
assertEquals(0, flowController.unconsumedBytes(connection().connectionStream()));
verify(mockKeepAliveManager).onDataReceived();
verify(streamListener, never()).closed(any(Status.class), any(RpcProgress.class),
any(Metadata.class));
verify(streamListener, never()).messagesAvailable(any(StreamListener.MessageProducer.class));

channelRead(headersFrame(STREAM_ID, new DefaultHttp2Headers().status(STATUS_OK)
.set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)));
verify(streamListener).headersRead(any(Metadata.class));
}

@Test
public void inboundShouldForwardToStream() throws Exception {
createStream();
Expand Down
Loading