Skip to content

Commit

Permalink
Fixes #10087 - Flaky EventsHandlerTest due to trailers. (#10093)
Browse files Browse the repository at this point in the history 
* Fixes #10087 - Flaky EventsHandlerTest due to trailers.

HTTP/2 trailers may arrive and be processed before the application reads request chunks.

Avoid the race condition by storing the trailers aside and returning them during reads.

Signed-off-by: Simone Bordet <[email protected]>
  • Loading branch information
@sbordet
sbordet authored Jul 13, 2023
1 parent 8c9d6e4 commit a08e953
Show file tree
Hide file tree
Showing 3 changed files with 114 additions and 48 deletions.
45 changes: 28 additions & 17 deletions ...ore/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Stream.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -371,32 +371,43 @@ private void onNewStream(Callback callback)

private void onHeaders(HeadersFrame frame, Callback callback)
{
boolean offered = false;
MetaData metaData = frame.getMetaData();
if (metaData.isRequest() || metaData.isResponse())
boolean isTrailer = !metaData.isRequest() && !metaData.isResponse();
if (isTrailer)
{
// In case of trailers, notify first and then offer EOF to
// avoid race conditions due to concurrent calls to readData().
boolean closed = updateClose(true, CloseState.Event.RECEIVED);
notifyHeaders(this, frame);
if (closed)
getSession().removeStream(this);
// Offer EOF in case the application calls readData() or demand().
offered = offer(Data.eof(getId()));
}
else
{
HttpFields fields = metaData.getHttpFields();
long length = -1;
if (fields != null && !HttpMethod.CONNECT.is(request.getMethod()))
length = fields.getLongField(HttpHeader.CONTENT_LENGTH);
dataLength = length;
}

boolean offered = false;
if (frame.isEndStream())
{
// Offer EOF for either the request, the response or the trailers
// in case the application calls readData() or demand().
offered = offer(Data.eof(getId()));
}
if (frame.isEndStream())
{
// Offer EOF for either the request or the response in
// case the application calls readData() or demand().
offered = offer(Data.eof(getId()));
}

// Requests are notified to a Session.Listener,
// here only handle responses and trailers.
if (metaData.isResponse() || !metaData.isRequest())
{
boolean closed = updateClose(frame.isEndStream(), CloseState.Event.RECEIVED);
notifyHeaders(this, frame);
if (closed)
getSession().removeStream(this);
// Requests are notified to a Session.Listener, here only notify responses.
if (metaData.isResponse())
{
boolean closed = updateClose(frame.isEndStream(), CloseState.Event.RECEIVED);
notifyHeaders(this, frame);
if (closed)
getSession().removeStream(this);
}
}

if (offered)
Expand Down
71 changes: 42 additions & 29 deletions ...tp2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HttpStreamOverHTTP2.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ public class HttpStreamOverHTTP2 implements HttpStream, HTTP2Channel.Server
private MetaData.Response _responseMetaData;
private TunnelSupport tunnelSupport;
private Content.Chunk _chunk;
private Content.Chunk _trailer;
private boolean committed;
private boolean _demand;
private boolean _expects100Continue;
Expand Down Expand Up @@ -150,37 +151,51 @@ public Content.Chunk read()
if (tunnelSupport != null)
return null;

while (true)
// Check if there already is a chunk, e.g. EOF.
Content.Chunk chunk;
try (AutoLock ignored = lock.lock())
{
Content.Chunk chunk;
try (AutoLock ignored = lock.lock())
{
chunk = _chunk;
_chunk = Content.Chunk.next(chunk);
}
if (chunk != null)
return chunk;

Stream.Data data = _stream.readData();
if (data == null)
return null;

// The data instance should be released after readData() above;
// the chunk is stored below for later use, so should be retained;
// the two actions cancel each other, no need to further retain or release.
chunk = createChunk(data);

// Some content is read, but the 100 Continue interim
// response has not been sent yet, then don't bother
// sending it later, as the client already sent the content.
if (_expects100Continue && chunk.hasRemaining())
_expects100Continue = false;
chunk = _chunk;
_chunk = Content.Chunk.next(chunk);
}
if (chunk != null)
return chunk;

Stream.Data data = _stream.readData();
if (data == null)
return null;

// Check if the trailers must be returned.
if (data.frame().isEndStream())
{
Content.Chunk trailer;
try (AutoLock ignored = lock.lock())
{
_chunk = chunk;
trailer = _trailer;
if (trailer != null)
{
_chunk = Content.Chunk.next(trailer);
return trailer;
}
}
}

// The data instance should be released after readData() above;
// the chunk is stored below for later use, so should be retained;
// the two actions cancel each other, no need to further retain or release.
chunk = createChunk(data);

// Some content is read, but the 100 Continue interim
// response has not been sent yet, then don't bother
// sending it later, as the client already sent the content.
if (_expects100Continue && chunk.hasRemaining())
_expects100Continue = false;

try (AutoLock ignored = lock.lock())
{
_chunk = Content.Chunk.next(chunk);
}
return chunk;
}

@Override
Expand All @@ -190,8 +205,7 @@ public void demand()
boolean demand = false;
try (AutoLock ignored = lock.lock())
{
// We may have a non-demanded chunk in case of trailers.
if (_chunk != null)
if (_chunk != null || _trailer != null)
notify = true;
else if (!_demand)
demand = _demand = true;
Expand Down Expand Up @@ -237,8 +251,7 @@ public Runnable onTrailer(HeadersFrame frame)
HttpFields trailers = frame.getMetaData().getHttpFields().asImmutable();
try (AutoLock ignored = lock.lock())
{
_demand = false;
_chunk = new Trailers(trailers);
_trailer = new Trailers(trailers);
}

if (LOG.isDebugEnabled())
Expand Down
46 changes: 44 additions & 2 deletions ...client-transports/src/test/java/org/eclipse/jetty/test/client/transport/TrailersTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,20 @@
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;

import org.eclipse.jetty.client.ContentResponse;
import org.eclipse.jetty.client.FutureResponseListener;
import org.eclipse.jetty.client.InputStreamResponseListener;
import org.eclipse.jetty.client.OutputStreamRequestContent;
import org.eclipse.jetty.client.StringRequestContent;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.io.Content;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Response;
import org.eclipse.jetty.util.Callback;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;

Expand All @@ -37,7 +42,6 @@ public class TrailersTest extends AbstractTest
{
@ParameterizedTest
@MethodSource("transportsNoFCGI")
@Tag("flaky") // https://github.com/eclipse/jetty.project/issues/9662
public void testTrailers(Transport transport) throws Exception
{
String trailerName = "Some-Trailer";
Expand Down Expand Up @@ -111,4 +115,42 @@ public void succeeded()
assertNotNull(responseTrailers);
assertEquals(trailerValue, responseTrailers.get(trailerName));
}

@ParameterizedTest
@MethodSource("transportsNoFCGI")
public void testTrailersWithDelayedRead(Transport transport) throws Exception
{
start(transport, new Handler.Abstract()
{
@Override
public boolean handle(Request request, Response response, Callback callback) throws Exception
{
// Do not read immediately, to cause the trailers to
// arrive at the server, especially in case of HTTP/2.
Thread.sleep(500);

HttpFields.Mutable trailers = HttpFields.build();
response.setTrailersSupplier(() -> trailers);
Content.copy(request, response, Response.newTrailersChunkProcessor(response), callback);

return true;
}
});

String content = "Some-Content";
String trailerName = "X-Trailer";
String trailerValue = "0xC0FFEE";
var request = client.newRequest(newURI(transport))
.method(HttpMethod.POST)
.headers(headers -> headers.put(HttpHeader.TRAILER, trailerName))
.body(new StringRequestContent(content))
.trailersSupplier(() -> HttpFields.build().put(trailerName, trailerValue));
FutureResponseListener listener = new FutureResponseListener(request);
request.send(listener);

ContentResponse response = listener.get(5, TimeUnit.SECONDS);
assertEquals(HttpStatus.OK_200, response.getStatus());
assertEquals(content, response.getContentAsString());
assertEquals(trailerValue, response.getTrailers().get(trailerName));
}
}

0 comments on commit a08e953

Please sign in to comment.