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鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Address maxConcurrentStreams violation on write timeout #3908

Merged
merged 10 commits into from
Nov 23, 2021
Merged

Address maxConcurrentStreams violation on write timeout #3908

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
225a684
add a simple reproducer
jrhee17 Nov 1, 2021
31f6571
don't remove response immediately on write timeout
jrhee17 Nov 1, 2021
e5d769f
Revert "add a simple reproducer"
jrhee17 Nov 1, 2021
fb719ef
fix npe in grpc client test
jrhee17 Nov 1, 2021
df43379
don't process headers/data/rst if res is closed
jrhee17 Nov 4, 2021
fbd4e8a
revise code documentation, revert test
jrhee17 Nov 6, 2021
8018da1
add a failing test
jrhee17 Nov 6, 2021
d1183d8
fix to send a goaway if should disconnect
jrhee17 Nov 7, 2021
7d207f9
fix test names, use long for seconds type directly
jrhee17 Nov 9, 2021
dca247c
prefer more explicit removeResponse method
jrhee17 Nov 23, 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
12 changes: 6 additions & 6 deletions core/src/main/java/com/linecorp/armeria/client/Http2ResponseDecoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,9 +96,9 @@ private void onWrapperCompleted(HttpResponseWrapper resWrapper, int id, @Nullabl
resWrapper.onSubscriptionCancelled(cause);

if (cause != null) {
// We are not closing the connection but just send a RST_STREAM,
// so we have to remove the response manually.
removeResponse(id);
ikhoon marked this conversation as resolved.
Show resolved Hide resolved
// Removing the response and decrementing {@code unfinishedResponses} isn't done immediately
// here. Instead, we rely on {@code Http2ResponseDecoder#onStreamClosed} to decrement
// `unfinishedResponses` to match the timing where netty decrements {@code numActiveStreams}.
Copy link
Member

@minwoox minwoox Nov 5, 2021 •
edited

Choose a reason for hiding this comment

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

How about?

Suggested change
// `unfinishedResponses` to match the timing where netty decrements {@code numActiveStreams}.
// `unfinishedResponses` after Netty decrements `numActiveStreams` in `DefaultHttp2Connection` so that `unfinishedResponses` is never greater than `numActiveStreams`.

(We don't have to use {@code ...} in the comment. 馃槈 )


// Reset the stream.
final int streamId = idToStreamId(id);
Expand Down Expand Up @@ -185,7 +185,7 @@ public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers
boolean endOfStream) throws Http2Exception {
keepAliveChannelRead();
final HttpResponseWrapper res = getResponse(streamIdToId(streamId), endOfStream);
if (res == null) {
if (res == null || !res.isOpen()) {
if (conn.streamMayHaveExisted(streamId)) {
if (logger.isDebugEnabled()) {
logger.debug("{} Received a late HEADERS frame for a closed stream: {}",
Expand Down Expand Up @@ -234,7 +234,7 @@ public int onDataRead(

final int dataLength = data.readableBytes();
final HttpResponseWrapper res = getResponse(streamIdToId(streamId), endOfStream);
if (res == null) {
if (res == null || !res.isOpen()) {
if (conn.streamMayHaveExisted(streamId)) {
if (logger.isDebugEnabled()) {
logger.debug("{} Received a late DATA frame for a closed stream: {}",
Expand Down Expand Up @@ -295,7 +295,7 @@ private boolean shouldSendGoAway() {
public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
keepAliveChannelRead();
final HttpResponseWrapper res = removeResponse(streamIdToId(streamId));
if (res == null) {
if (res == null || !res.isOpen()) {
if (conn.streamMayHaveExisted(streamId)) {
if (logger.isDebugEnabled()) {
logger.debug("{} Received a late RST_STREAM frame for a closed stream: {}",
Expand Down
1 change: 1 addition & 0 deletions grpc/src/test/java/com/linecorp/armeria/client/grpc/GrpcClientTimeoutTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ void clientTimeout() throws InterruptedException {
assertThat(loggingEventCaptor.getAllValues()).noneMatch(event -> {
return event.getLevel() == Level.WARN &&
event.getThrowableProxy() != null &&
event.getThrowableProxy().getMessage() != null &&
Copy link
Member

Choose a reason for hiding this comment

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

revert?

event.getThrowableProxy().getMessage().contains("call already closed");
});
}
Expand Down