Skip to content

Commit

Permalink
Fixes #8678 - Jetty client receives GO_AWAY and continue to send traf…
Browse files Browse the repository at this point in the history
…fic on same connection

* Now upon receiving the GOAWAY, the connection is removed from the pool, so it cannot be used by new requests.
* HTTP2Session.removeStream() now happens _after_ notifying HEADERS and DATA events, although the Stream state change still happens before.
This is necessary to avoid that a "close" event is notified before a "headers" or "data" event.

With these changes, the race window of a client acquiring a connection while the server is closing it is reduced, but it is impossible to close it completely.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
  • Loading branch information
sbordet committed Nov 12, 2022
1 parent 12041da commit fbc23ac
Show file tree
Hide file tree
Showing 8 changed files with 208 additions and 28 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,10 @@ public void onHeaders(HeadersFrame frame)
else
{
stream.process(frame, Callback.NOOP);
if (stream.updateClose(frame.isEndStream(), CloseState.Event.RECEIVED))
removeStream(stream);
boolean closed = stream.updateClose(frame.isEndStream(), CloseState.Event.RECEIVED);
notifyHeaders(stream, frame);
if (closed)
removeStream(stream);
}
}
else
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2139,12 +2139,8 @@ private void newLocalStream(IStream.FrameList frameList, Promise<Stream> promise

private Stream newUpgradeStream(HeadersFrame frame, Stream.Listener listener, Consumer<Throwable> failFn)
{
int streamId;
try (AutoLock ignored = lock.lock())
{
streamId = localStreamIds.getAndAdd(2);
HTTP2Session.this.onStreamCreated(streamId);
}
int streamId = localStreamIds.getAndAdd(2);
HTTP2Session.this.onStreamCreated(streamId);
IStream stream = HTTP2Session.this.createLocalStream(streamId, (MetaData.Request)frame.getMetaData(), x ->
{
HTTP2Session.this.onStreamDestroyed(streamId);
Expand All @@ -2160,30 +2156,33 @@ private Stream newUpgradeStream(HeadersFrame frame, Stream.Listener listener, Co

private boolean newRemoteStream(int streamId)
{
boolean created = false;
try (AutoLock ignored = lock.lock())
{
switch (closed)
{
case NOT_CLOSED:
{
HTTP2Session.this.onStreamCreated(streamId);
return true;
created = true;
break;
}
case LOCALLY_CLOSED:
{
// SPEC: streams larger than GOAWAY's lastStreamId are dropped.
if (streamId <= goAwaySent.getLastStreamId())
{
// Allow creation of streams that may have been in-flight.
HTTP2Session.this.onStreamCreated(streamId);
return true;
created = true;
}
return false;
break;
}
default:
return false;
break;
}
}
if (created)
HTTP2Session.this.onStreamCreated(streamId);
return created;
}

private void push(PushPromiseFrame frame, Promise<Stream> promise, Stream.Listener listener)
Expand Down Expand Up @@ -2244,14 +2243,16 @@ private MetaData.Request extractMetaDataRequest(StreamFrame frame)
private int reserveSlot(Slot slot, int streamId, Consumer<Throwable> fail)
{
Throwable failure = null;
boolean reserved = false;
try (AutoLock ignored = lock.lock())
{
// SPEC: cannot create new streams after receiving a GOAWAY.
if (closed == CloseState.NOT_CLOSED)
{
if (streamId <= 0)
{
streamId = localStreamIds.getAndAdd(2);
HTTP2Session.this.onStreamCreated(streamId);
reserved = true;
}
slots.offer(slot);
}
Expand All @@ -2263,9 +2264,16 @@ private int reserveSlot(Slot slot, int streamId, Consumer<Throwable> fail)
}
}
if (failure == null)
{
if (reserved)
HTTP2Session.this.onStreamCreated(streamId);
return streamId;
fail.accept(failure);
return 0;
}
else
{
fail.accept(failure);
return 0;
}
}

private void freeSlot(Slot slot, int streamId)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -536,9 +536,10 @@ private void processData()
dataEntry = dataQueue.poll();
}
DataFrame frame = dataEntry.frame;
if (updateClose(frame.isEndStream(), CloseState.Event.RECEIVED))
session.removeStream(this);
boolean closed = updateClose(frame.isEndStream(), CloseState.Event.RECEIVED);
notifyDataDemanded(this, frame, dataEntry.callback);
if (closed)
session.removeStream(this);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,6 @@ public HeadersFrame withStreamId(int streamId)
@Override
public String toString()
{
return String.format("%s#%d{end=%b}%s", super.toString(), getStreamId(), endStream,
priority == null ? "" : String.format("+%s", priority));
return String.format("%s#%d[end=%b,{%s},priority=%s]", super.toString(), getStreamId(), isEndStream(), getMetaData(), getPriority());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -81,12 +81,22 @@ protected HttpConnectionOverHTTP2 newHttpConnection(HttpDestination destination,
return new HttpConnectionOverHTTP2(destination, session);
}

@Override
public void onGoAway(Session session, GoAwayFrame frame)
{
if (failConnectionPromise(new ClosedChannelException()))
return;
HttpConnectionOverHTTP2 connection = getConnection();
if (connection != null)
connection.remove();
}

@Override
public void onClose(Session session, GoAwayFrame frame)
{
if (failConnectionPromise(new ClosedChannelException()))
return;
HttpConnectionOverHTTP2 connection = this.connection.getReference();
HttpConnectionOverHTTP2 connection = getConnection();
if (connection != null)
onClose(connection, frame);
}
Expand All @@ -103,7 +113,7 @@ public boolean onIdleTimeout(Session session)
TimeoutException failure = new TimeoutException("Idle timeout expired: " + idleTimeout + " ms");
if (failConnectionPromise(failure))
return true;
HttpConnectionOverHTTP2 connection = this.connection.getReference();
HttpConnectionOverHTTP2 connection = getConnection();
if (connection != null)
return connection.onIdleTimeout(idleTimeout, failure);
return true;
Expand All @@ -114,7 +124,7 @@ public void onFailure(Session session, Throwable failure)
{
if (failConnectionPromise(failure))
return;
HttpConnectionOverHTTP2 connection = this.connection.getReference();
HttpConnectionOverHTTP2 connection = getConnection();
if (connection != null)
connection.close(failure);
}
Expand All @@ -126,4 +136,9 @@ private boolean failConnectionPromise(Throwable failure)
httpConnectionPromise().failed(failure);
return result;
}

private HttpConnectionOverHTTP2 getConnection()
{
return connection.getReference();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,11 @@ public boolean onIdleTimeout(long idleTimeout, Throwable failure)
return false;
}

void remove()
{
getHttpDestination().remove(this);
}

@Override
public void close()
{
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//

package org.eclipse.jetty.http2.client.http;

import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.eclipse.jetty.client.api.Response;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.http.MetaData;
import org.eclipse.jetty.http2.ErrorCode;
import org.eclipse.jetty.http2.ISession;
import org.eclipse.jetty.http2.api.Stream;
import org.eclipse.jetty.http2.api.server.ServerSessionListener;
import org.eclipse.jetty.http2.frames.HeadersFrame;
import org.eclipse.jetty.util.Callback;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class GoAwayTest extends AbstractTest
{
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testConnectionIsRemovedFromPoolOnGracefulGoAwayReceived(boolean graceful) throws Exception
{
long timeout = 5000;
AtomicReference<Response> responseRef = new AtomicReference<>();
CountDownLatch responseLatch = new CountDownLatch(1);
start(new ServerSessionListener.Adapter()
{
private Stream goAwayStream;

@Override
public Stream.Listener onNewStream(Stream stream, HeadersFrame frame)
{
MetaData.Request request = (MetaData.Request)frame.getMetaData();
String path = request.getURI().getPath();

if ("/prime".equals(path))
{
respond(stream);
}
else if ("/goaway".equals(path))
{
try
{
goAwayStream = stream;

if (graceful)
{
// Send to the client a graceful GOAWAY.
((ISession)stream.getSession()).shutdown();
}
else
{
// Send to the client a non-graceful GOAWAY.
stream.getSession().close(ErrorCode.ENHANCE_YOUR_CALM_ERROR.code, null, Callback.NOOP);
}

// Wait for the client to receive the GOAWAY.
Thread.sleep(1000);

// This request will be performed on a different connection.
client.newRequest("localhost", connector.getLocalPort())
.path("/after")
.timeout(timeout / 2, TimeUnit.MILLISECONDS)
.send(result ->
{
responseRef.set(result.getResponse());
responseLatch.countDown();
});
}
catch (Exception x)
{
throw new RuntimeException(x);
}
}
else if ("/after".equals(path))
{
// Wait for the /after request to arrive to the server
// before answering to the /goaway request.
// The /goaway request must succeed because it's in
// flight and seen by the server when the GOAWAY happens,
// so it will be completed before closing the connection.
respond(goAwayStream);
respond(stream);
}
return null;
}

private void respond(Stream stream)
{
long remotePort = ((InetSocketAddress)stream.getSession().getRemoteSocketAddress()).getPort();
HttpFields responseHeaders = HttpFields.build().putLongField("X-Remote-Port", remotePort);
MetaData.Response response = new MetaData.Response(HttpVersion.HTTP_2, HttpStatus.OK_200, responseHeaders);
stream.headers(new HeadersFrame(stream.getId(), response, null, true));
}
});

Response response = client.newRequest("localhost", connector.getLocalPort())
.path("/prime")
.timeout(timeout, TimeUnit.MILLISECONDS)
.send();
assertEquals(HttpStatus.OK_200, response.getStatus());
long primePort = response.getHeaders().getLongField("X-Remote-Port");

response = client.newRequest("localhost", connector.getLocalPort())
.path("/goaway")
.timeout(timeout, TimeUnit.MILLISECONDS)
.send();
assertEquals(HttpStatus.OK_200, response.getStatus());
long goAwayPort = response.getHeaders().getLongField("X-Remote-Port");
assertEquals(primePort, goAwayPort);

assertTrue(responseLatch.await(timeout, TimeUnit.MILLISECONDS));
response = responseRef.get();
assertNotNull(response);
assertEquals(HttpStatus.OK_200, response.getStatus());
// The /after request must happen on a different port
// because the first connection has been removed from the pool.
long afterPort = response.getHeaders().getLongField("X-Remote-Port");
assertNotEquals(primePort, afterPort);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -110,10 +110,11 @@ public void onHeaders(HeadersFrame frame)
}

stream.process(frame, Callback.NOOP);
if (stream.updateClose(frame.isEndStream(), CloseState.Event.RECEIVED))
removeStream(stream);
boolean closed = stream.updateClose(frame.isEndStream(), CloseState.Event.RECEIVED);
Stream.Listener listener = notifyNewStream(stream, frame);
stream.setListener(listener);
if (closed)
removeStream(stream);
}
}
}
Expand All @@ -132,9 +133,10 @@ else if (metaData.isResponse())
if (stream != null)
{
stream.process(frame, Callback.NOOP);
if (stream.updateClose(frame.isEndStream(), CloseState.Event.RECEIVED))
removeStream(stream);
boolean closed = stream.updateClose(frame.isEndStream(), CloseState.Event.RECEIVED);
notifyHeaders(stream, frame);
if (closed)
removeStream(stream);
}
else
{
Expand Down

0 comments on commit fbc23ac

Please sign in to comment.