Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixes #12227 - Improve HttpConnection buffer recycling. (#12237) 

In this PR, the responsibility to release the buffers is in 2 methods: onFillable() (called when network data is available, and to process the next request) and parseAndFillForContent() (called from Request.read()).

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
Co-authored-by: gregw <gregw@webtide.com>
This commit is contained in:
Simone Bordet 2024-09-06 07:11:29 +03:00 committed by GitHub
parent f0a1db100f
commit 6887435773
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 210 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java
View File

@ -482,7 +482,8 @@ public class HttpChannelState implements HttpChannel, Components
}
}
// Consume content as soon as possible to open any flow control window.
// Consume content as soon as possible to open any
// flow control window and release any request buffer.
Throwable unconsumed = stream.consumeAvailable();
if (unconsumed != null && LOG.isDebugEnabled())
LOG.debug("consuming content during error {}", unconsumed.toString());

268
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpConnection.java
View File

@ -19,6 +19,7 @@ import java.nio.channels.WritePendingException;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
@ -103,7 +104,7 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
private final LongAdder bytesOut = new LongAdder();
private final AtomicBoolean _handling = new AtomicBoolean(false);
private final HttpFields.Mutable _headerBuilder = HttpFields.build();
private volatile RetainableByteBuffer _retainableByteBuffer;
private volatile RetainableByteBuffer _requestBuffer;
private HttpFields.Mutable _trailers;
private Runnable _onRequest;
private long _requests;
@ -319,70 +320,65 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
public ByteBuffer onUpgradeFrom()
{
if (isRequestBufferEmpty())
{
releaseRequestBuffer();
return null;
}
ByteBuffer unconsumed = ByteBuffer.allocateDirect(_retainableByteBuffer.remaining());
unconsumed.put(_retainableByteBuffer.getByteBuffer());
ByteBuffer unconsumed = ByteBuffer.allocateDirect(_requestBuffer.remaining());
unconsumed.put(_requestBuffer.getByteBuffer());
unconsumed.flip();
releaseRequestBuffer();
return unconsumed;
}
@Override
public void onUpgradeTo(ByteBuffer buffer)
{
BufferUtil.append(getRequestBuffer(), buffer);
ensureRequestBuffer();
BufferUtil.append(_requestBuffer.getByteBuffer(), buffer);
}
void releaseRequestBuffer()
{
if (_retainableByteBuffer != null && !_retainableByteBuffer.hasRemaining())
private void releaseRequestBuffer()
{
if (LOG.isDebugEnabled())
LOG.debug("releaseRequestBuffer {}", this);
RetainableByteBuffer buffer = _retainableByteBuffer;
_retainableByteBuffer = null;
if (!buffer.release())
throw new IllegalStateException("unreleased buffer " + buffer);
}
LOG.debug("releasing request buffer {} {}", _requestBuffer, this);
_requestBuffer.release();
_requestBuffer = null;
}
private ByteBuffer getRequestBuffer()
private void ensureRequestBuffer()
{
if (_retainableByteBuffer == null)
_retainableByteBuffer = _bufferPool.acquire(getInputBufferSize(), isUseInputDirectByteBuffers());
return _retainableByteBuffer.getByteBuffer();
if (_requestBuffer == null)
{
_requestBuffer = _bufferPool.acquire(getInputBufferSize(), isUseInputDirectByteBuffers());
if (LOG.isDebugEnabled())
LOG.debug("request buffer acquired {} {}", _requestBuffer, this);
}
}
public boolean isRequestBufferEmpty()
{
return _retainableByteBuffer == null || !_retainableByteBuffer.hasRemaining();
return _requestBuffer == null || !_requestBuffer.hasRemaining();
}
@Override
public void onFillable()
{
if (LOG.isDebugEnabled())
LOG.debug(">>onFillable enter {} {} {}", this, _httpChannel, _retainableByteBuffer);
LOG.debug("onFillable enter {} {} {}", _httpChannel, _requestBuffer, this);
HttpConnection last = setCurrentConnection(this);
try
{
ensureRequestBuffer();
// We must loop until we fill -1 or there is an async pause in handling.
// Note that the endpoint might already be closed in some special circumstances.
while (true)
{
// Fill the request buffer (if needed).
int filled = fillRequestBuffer();
if (LOG.isDebugEnabled())
LOG.debug("onFillable filled {} {} {} {}", filled, this, _httpChannel, _retainableByteBuffer);
LOG.debug("onFillable filled {} {} {} {}", filled, _httpChannel, _requestBuffer, this);
if (filled < 0 && getEndPoint().isOutputShutdown())
close();
// Parse the request buffer.
boolean handle = parseRequestBuffer();
// There could be a connection upgrade before handling
@ -390,52 +386,66 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
// If there was a connection upgrade, the other
// connection took over, nothing more to do here.
if (getEndPoint().getConnection() != this)
{
releaseRequestBuffer();
break;
}
// Handle channel event. This will only be true when the headers of a request have been received.
// The headers of a request have been received.
if (handle)
{
Request request = _httpChannel.getRequest();
if (LOG.isDebugEnabled())
LOG.debug("HANDLE {} {}", request, this);
// handle the request by running the task obtained from onRequest
// Handle the request by running the task.
_handling.set(true);
Runnable onRequest = _onRequest;
_onRequest = null;
onRequest.run();
// If the _handling boolean has already been CaS'd to false, then stream is completed and we are no longer
// handling, so the caller can continue to fill and parse more connections. If it is still true, then some
// thread is still handling the request and they will need to organize more filling and parsing once complete.
// If the CaS succeeds, then some thread is still handling the request.
// If the CaS fails, then stream is completed, we are no longer handling,
// so the caller can continue to fill and parse more connections.
if (_handling.compareAndSet(true, false))
{
if (LOG.isDebugEnabled())
LOG.debug("request !complete {} {}", request, this);
LOG.debug("request !complete {} {} {}", request, _requestBuffer, this);
// Cannot release the request buffer here, because the
// application may read concurrently from another thread.
// The request buffer will be released by the application
// reading the request content, or by the implementation
// trying to consume the request content.
break;
}
// If the request is complete, but has been upgraded, then break
// If there was an upgrade, release and return.
if (getEndPoint().getConnection() != this)
{
if (LOG.isDebugEnabled())
LOG.debug("upgraded {} -> {}", this, getEndPoint().getConnection());
releaseRequestBuffer();
break;
}
}
else if (filled == 0)
{
assert isRequestBufferEmpty();
releaseRequestBuffer();
fillInterested();
break;
}
else if (filled < 0)
{
assert isRequestBufferEmpty();
releaseRequestBuffer();
getEndPoint().shutdownOutput();
break;
}
else if (_requestHandler._failure != null)
{
// There was an error, don't fill more.
break;
}
else if (filled == 0)
{
fillInterested();
releaseRequestBuffer();
break;
}
}
@ -446,12 +456,9 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
{
if (LOG.isDebugEnabled())
LOG.debug("caught exception {} {}", this, _httpChannel, x);
if (_retainableByteBuffer != null)
{
_retainableByteBuffer.clear();
if (_requestBuffer != null)
releaseRequestBuffer();
}
}
finally
{
getEndPoint().close(x);
@ -461,7 +468,7 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
{
setCurrentConnection(last);
if (LOG.isDebugEnabled())
LOG.debug("<<onFillable exit {} {} {}", this, _httpChannel, _retainableByteBuffer);
LOG.debug("onFillable exit {} {} {}", _httpChannel, _requestBuffer, this);
}
}
@ -471,70 +478,63 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
*/
void parseAndFillForContent()
{
// Defensive check to avoid an infinite select/wakeup/fillAndParseForContent/wait loop
// in case the parser was mistakenly closed and the connection was not aborted.
if (_parser.isTerminated())
{
_requestHandler.messageComplete();
return;
}
ensureRequestBuffer();
// When fillRequestBuffer() is called, it must always be followed by a parseRequestBuffer() call otherwise this method
// doesn't trigger EOF/earlyEOF which breaks AsyncRequestReadTest.testPartialReadThenShutdown().
// This loop was designed by a committee and voted by a majority.
while (_parser.inContentState())
{
if (parseRequestBuffer())
break;
// Re-check the parser state after parsing to avoid filling,
// otherwise fillRequestBuffer() would acquire a ByteBuffer
// that may be leaked.
if (_parser.inContentState() && fillRequestBuffer() <= 0)
if (!_parser.inContentState())
{
// The request is complete, and we are going to re-enter onFillable(),
// because either A: the request/response was completed synchronously
// so the onFillable() thread will loop, or B: the request/response
// was completed asynchronously, and the HttpStreamOverHTTP1 dispatches
// a call to onFillable() to process the next request.
// Therefore, there is no need to release the request buffer here,
// also because the buffer may contain pipelined requests.
break;
}
assert !_requestBuffer.hasRemaining();
if (_requestBuffer.isRetained())
{
// The application has retained the content chunks,
// reacquire the buffer to avoid overwriting the content.
releaseRequestBuffer();
ensureRequestBuffer();
}
int filled = fillRequestBuffer();
if (filled <= 0)
{
releaseRequestBuffer();
break;
}
}
}
private int fillRequestBuffer()
{
if (_retainableByteBuffer != null && _retainableByteBuffer.isRetained())
{
// TODO this is almost certainly wrong
RetainableByteBuffer newBuffer = _bufferPool.acquire(getInputBufferSize(), isUseInputDirectByteBuffers());
if (LOG.isDebugEnabled())
LOG.debug("replace buffer {} <- {} in {}", _retainableByteBuffer, newBuffer, this);
_retainableByteBuffer.release();
_retainableByteBuffer = newBuffer;
}
if (!isRequestBufferEmpty())
return _retainableByteBuffer.remaining();
return _requestBuffer.remaining();
// Get a buffer
// We are not in a race here for the request buffer as we have not yet received a request,
// so there are not any possible legal threads calling #parseContent or #completed.
ByteBuffer requestBuffer = getRequestBuffer();
// fill
try
{
ByteBuffer requestBuffer = _requestBuffer.getByteBuffer();
int filled = getEndPoint().fill(requestBuffer);
if (filled == 0) // Do a retry on fill 0 (optimization for SSL connections)
filled = getEndPoint().fill(requestBuffer);
if (LOG.isDebugEnabled())
LOG.debug("{} filled {} {}", this, filled, _retainableByteBuffer);
LOG.debug("filled {} {} {}", filled, _requestBuffer, this);
if (filled > 0)
{
bytesIn.add(filled);
}
else
{
if (filled < 0)
else if (filled < 0)
_parser.atEOF();
releaseRequestBuffer();
}
return filled;
}
@ -543,11 +543,6 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
if (LOG.isDebugEnabled())
LOG.debug("Unable to fill from endpoint {}", getEndPoint(), x);
_parser.atEOF();
if (_retainableByteBuffer != null)
{
_retainableByteBuffer.clear();
releaseRequestBuffer();
}
return -1;
}
}
@ -555,19 +550,15 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
private boolean parseRequestBuffer()
{
if (LOG.isDebugEnabled())
LOG.debug("{} parse {}", this, _retainableByteBuffer);
LOG.debug("parse {} {}", _requestBuffer, this);
if (_parser.isTerminated())
throw new RuntimeIOException("Parser is terminated");
boolean handle = _parser.parseNext(_retainableByteBuffer == null ? BufferUtil.EMPTY_BUFFER : _retainableByteBuffer.getByteBuffer());
boolean handle = _parser.parseNext(_requestBuffer.getByteBuffer());
if (LOG.isDebugEnabled())
LOG.debug("{} parsed {} {}", this, handle, _parser);
// recycle buffer ?
if (_retainableByteBuffer != null && !_retainableByteBuffer.isRetained())
releaseRequestBuffer();
LOG.debug("parsed {} {} {} {}", handle, _parser, _requestBuffer, this);
return handle;
}
@ -624,19 +615,6 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
getExecutor().execute(this);
}
@Override
public void onClose(Throwable cause)
{
// TODO: do we really need to do this?
// This event is fired really late, sendCallback should already be failed at this point.
// Revisit whether we still need IteratingCallback.close().
if (cause == null)
_sendCallback.close();
else
_sendCallback.abort(cause);
super.onClose(cause);
}
@Override
public void run()
{
@ -969,14 +947,14 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
public boolean content(ByteBuffer buffer)
{
HttpStreamOverHTTP1 stream = _stream.get();
if (stream == null || stream._chunk != null || _retainableByteBuffer == null)
if (stream == null || stream._chunk != null || _requestBuffer == null)
throw new IllegalStateException();
if (LOG.isDebugEnabled())
LOG.debug("content {}/{} for {}", BufferUtil.toDetailString(buffer), _retainableByteBuffer, HttpConnection.this);
LOG.debug("content {}/{} for {}", BufferUtil.toDetailString(buffer), _requestBuffer, HttpConnection.this);
_retainableByteBuffer.retain();
stream._chunk = Content.Chunk.asChunk(buffer, false, _retainableByteBuffer);
_requestBuffer.retain();
stream._chunk = Content.Chunk.asChunk(buffer, false, _requestBuffer);
return true;
}
@ -1520,6 +1498,11 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
if (LOG.isDebugEnabled())
LOG.debug("abort due to pending read {} {} ", this, getEndPoint());
abort(new IOException("Pending read in onCompleted"));
_httpChannel.recycle();
_parser.reset();
_generator.reset();
if (!_handling.compareAndSet(true, false))
resume();
return;
}
@ -1529,6 +1512,8 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
_httpChannel.recycle();
_parser.close();
_generator.reset();
if (!_handling.compareAndSet(true, false))
releaseRequestBuffer();
return;
}
@ -1560,40 +1545,7 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
if (LOG.isDebugEnabled())
LOG.debug("non-current completion {}", this);
// If we are looking for the next request
if (_parser.isStart())
{
// if the buffer is empty
if (isRequestBufferEmpty())
{
// look for more data
fillInterested();
}
// else if we are still running
else if (getConnector().isRunning())
{
// Dispatched to handle a pipelined request
try
{
getExecutor().execute(HttpConnection.this);
}
catch (RejectedExecutionException e)
{
if (getConnector().isRunning())
LOG.warn("Failed dispatch of {}", this, e);
else
LOG.trace("IGNORED", e);
getEndPoint().close();
}
}
else
{
getEndPoint().close();
}
}
// else the parser must be closed, so seek the EOF if we are still open
else if (getEndPoint().isOpen())
fillInterested();
resume();
}
@Override
@ -1609,6 +1561,30 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab
if (LOG.isDebugEnabled())
LOG.debug("aborting", x);
abort(x);
_httpChannel.recycle();
_parser.reset();
_generator.reset();
if (!_handling.compareAndSet(true, false))
resume();
}
private void resume()
{
try
{
if (LOG.isDebugEnabled())
LOG.debug("Resuming onFillable() {}", HttpConnection.this);
// Dispatch to handle pipelined requests.
Request request = _httpChannel.getRequest();
Executor executor = request == null ? getExecutor() : request.getComponents().getExecutor();
executor.execute(HttpConnection.this);
}
catch (RejectedExecutionException x)
{
getEndPoint().close(x);
// Resume by running, to release the request buffer.
run();
}
}
private void abort(Throwable failure)

86
jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/HttpServerTestBase.java
View File

@ -60,7 +60,7 @@ import org.hamcrest.Matchers;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.junit.jupiter.params.provider.CsvSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -68,7 +68,6 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.endsWith;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
@ -1869,8 +1868,8 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture
}
@ParameterizedTest
@ValueSource(booleans = {false /* TODO, true */})
public void testHoldContent(boolean close) throws Exception
@CsvSource({"false,false", "false,true", "true,false", "true,true"})
public void testHoldContent(boolean close, boolean pipeline) throws Exception
{
Queue<Content.Chunk> contents = new ConcurrentLinkedQueue<>();
final int bufferSize = 1024;
@ -1881,6 +1880,15 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture
@Override
public boolean handle(Request request, Response response, Callback callback) throws Exception
{
if (LOG.isDebugEnabled())
LOG.debug("Handling request: {}", request);
if ("GET".equals(request.getMethod()))
{
response.setStatus(200);
callback.succeeded();
return true;
}
request.getConnectionMetaData().getConnection().addEventListener(new Connection.Listener()
{
@Override
@ -1889,10 +1897,12 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture
closed.countDown();
}
});
while (true)
{
Content.Chunk chunk = request.read();
if (LOG.isDebugEnabled())
LOG.debug("read: {}", chunk);
if (chunk == null)
{
try (Blocker.Runnable blocker = Blocker.runnable())
@ -1902,81 +1912,111 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture
continue;
}
}
if (chunk.hasRemaining())
contents.add(chunk);
else
chunk.release();
if (chunk.isLast())
break;
}
response.setStatus(200);
if (close)
{
LOG.info("Closing {}", request.getConnectionMetaData().getConnection().getEndPoint());
request.getConnectionMetaData().getConnection().getEndPoint().close();
}
callback.succeeded();
return true;
}
});
byte[] chunk = new byte[bufferSize / 2];
Arrays.fill(chunk, (byte)'X');
try (Socket client = newSocket(_serverURI.getHost(), _serverURI.getPort()))
{
OutputStream os = client.getOutputStream();
BufferedOutputStream out = new BufferedOutputStream(os, bufferSize);
out.write(("""
String request = """
POST / HTTP/1.1\r
Host: localhost\r
Connection: close\r
Connection: %s\r
Transfer-Encoding: chunked\r
\r
""").getBytes(StandardCharsets.ISO_8859_1));
""".formatted(pipeline ? "other" : "close");
if (LOG.isDebugEnabled())
LOG.debug("raw request {}", request);
out.write(request.getBytes(StandardCharsets.ISO_8859_1));
// single chunk
out.write((Integer.toHexString(chunk.length) + "\r\n").getBytes(StandardCharsets.ISO_8859_1));
out.write(chunk);
out.write("\r\n".getBytes(StandardCharsets.ISO_8859_1));
out.flush();
// double chunk (will overflow)
// double chunk (will overflow bufferSize)
out.write((Integer.toHexString(chunk.length * 2) + "\r\n").getBytes(StandardCharsets.ISO_8859_1));
out.write(chunk);
out.write(chunk);
out.write("\r\n".getBytes(StandardCharsets.ISO_8859_1));
out.flush();
// single chunk and end chunk
out.write((Integer.toHexString(chunk.length) + "\r\n").getBytes(StandardCharsets.ISO_8859_1));
out.write(chunk);
out.write("\r\n0\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1));
// single chunk and end chunk plus optional pipelined request
ByteBuffer last = BufferUtil.allocate(4096);
BufferUtil.append(last, (Integer.toHexString(chunk.length) + "\r\n").getBytes(StandardCharsets.ISO_8859_1));
BufferUtil.append(last, chunk);
BufferUtil.append(last, "\r\n0\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1));
if (pipeline)
BufferUtil.append(last, "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1));
if (LOG.isDebugEnabled())
LOG.debug("last {}", BufferUtil.toString(last));
out.write(BufferUtil.toArray(last));
out.flush();
// check the response
if (!close)
if (close)
{
assertThat(client.getInputStream().read(), equalTo(-1));
}
else
{
HttpTester.Response response = HttpTester.parseResponse(client.getInputStream());
assertNotNull(response);
assertThat(response.getStatus(), is(200));
if (pipeline)
{
response = HttpTester.parseResponse(client.getInputStream());
assertNotNull(response);
assertThat(response.getStatus(), is(200));
}
}
}
assertTrue(closed.await(10, TimeUnit.SECONDS));
long total = contents.stream().mapToLong(Content.Chunk::remaining).sum();
assertThat(total, equalTo(chunk.length * 4L));
ByteBufferPool rbbp = _connector.getByteBufferPool();
if (rbbp instanceof ArrayByteBufferPool pool)
{
long buffersBeforeRelease = pool.getAvailableDirectByteBufferCount() + pool.getAvailableHeapByteBufferCount();
if (LOG.isDebugEnabled())
{
LOG.debug("pool {}", pool);
contents.stream().map(Content.Chunk::toString).forEach(LOG::debug);
}
contents.forEach(Content.Chunk::release);
Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() ->
{
if (LOG.isDebugEnabled())
{
LOG.debug("pool {}", pool);
contents.stream().map(Content.Chunk::toString).forEach(LOG::debug);
}
long buffersAfterRelease = pool.getAvailableDirectByteBufferCount() + pool.getAvailableHeapByteBufferCount();
assertThat(buffersAfterRelease, greaterThan(buffersBeforeRelease));
return buffersAfterRelease > buffersBeforeRelease;
});
assertThat(pool.getAvailableDirectMemory() + pool.getAvailableHeapMemory(), greaterThanOrEqualTo(chunk.length * 4L));
}
else

6
jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/RequestListenersTest.java
View File

@ -191,11 +191,13 @@ public class RequestListenersTest
Content-Length: 1
Connection: close
""", 2000 * idleTimeout, TimeUnit.MILLISECONDS));
""", 2 * idleTimeout, TimeUnit.MILLISECONDS));
int expectedStatus = succeedCallback ? HttpStatus.OK_200 : HttpStatus.INTERNAL_SERVER_ERROR_500;
assertEquals(expectedStatus, response.getStatus());
assertThat(failureLatch.await(idleTimeout + 500, TimeUnit.MILLISECONDS), is(failIdleTimeout && !succeedCallback));
// The failure listener is never invoked because completing the callback
// produces a response that completes the stream so the failure is ignored.
assertThat(failureLatch.await(idleTimeout + 500, TimeUnit.MILLISECONDS), is(false));
}
@ParameterizedTest

26
jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/HttpClientIdleTimeoutTest.java
View File

@ -15,6 +15,8 @@ package org.eclipse.jetty.test.client.transport;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.client.ContentResponse;
import org.eclipse.jetty.client.StringRequestContent;
@ -26,6 +28,8 @@ import org.eclipse.jetty.util.Callback;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@ -38,17 +42,17 @@ public class HttpClientIdleTimeoutTest extends AbstractTest
public void testClientIdleTimeout(Transport transport) throws Exception
{
long serverIdleTimeout = idleTimeout * 2;
CountDownLatch serverIdleTimeoutLatch = new CountDownLatch(1);
AtomicReference<Callback> serverCallbackRef = new AtomicReference<>();
start(transport, new Handler.Abstract()
{
@Override
public boolean handle(Request request, Response response, Callback callback)
{
// Do not succeed the callback if it's a timeout request.
if (!Request.getPathInContext(request).equals("/timeout"))
callback.succeeded();
if (Request.getPathInContext(request).equals("/timeout"))
request.addFailureListener(x -> serverCallbackRef.set(callback));
else
request.addFailureListener(x -> serverIdleTimeoutLatch.countDown());
callback.succeeded();
return true;
}
});
@ -75,7 +79,8 @@ public class HttpClientIdleTimeoutTest extends AbstractTest
assertEquals(HttpStatus.OK_200, response.getStatus());
// Wait for the server's idle timeout to trigger to give it a chance to clean up its resources.
assertTrue(serverIdleTimeoutLatch.await(2 * serverIdleTimeout, TimeUnit.MILLISECONDS));
Callback callback = await().atMost(2 * serverIdleTimeout, TimeUnit.MILLISECONDS).until(serverCallbackRef::get, notNullValue());
callback.failed(new TimeoutException());
}
@ParameterizedTest
@ -83,17 +88,17 @@ public class HttpClientIdleTimeoutTest extends AbstractTest
public void testRequestIdleTimeout(Transport transport) throws Exception
{
long serverIdleTimeout = idleTimeout * 2;
CountDownLatch serverIdleTimeoutLatch = new CountDownLatch(1);
AtomicReference<Callback> serverCallbackRef = new AtomicReference<>();
start(transport, new Handler.Abstract()
{
@Override
public boolean handle(Request request, Response response, Callback callback) throws Exception
{
// Do not succeed the callback if it's a timeout request.
if (!Request.getPathInContext(request).equals("/timeout"))
callback.succeeded();
if (Request.getPathInContext(request).equals("/timeout"))
request.addFailureListener(x -> serverCallbackRef.set(callback));
else
request.addFailureListener(x -> serverIdleTimeoutLatch.countDown());
callback.succeeded();
return true;
}
});
@ -120,7 +125,8 @@ public class HttpClientIdleTimeoutTest extends AbstractTest
assertEquals(HttpStatus.OK_200, response.getStatus());
// Wait for the server's idle timeout to trigger to give it a chance to clean up its resources.
assertTrue(serverIdleTimeoutLatch.await(2 * serverIdleTimeout, TimeUnit.MILLISECONDS));
Callback callback = await().atMost(2 * serverIdleTimeout, TimeUnit.MILLISECONDS).until(serverCallbackRef::get, notNullValue());
callback.failed(new TimeoutException());
}
@ParameterizedTest