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

Issue #2796 - Max local stream count exceeded when request fails. 

Bound release of the channel to stream close event.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
This commit is contained in:
Simone Bordet 2018-10-11 15:56:47 +02:00
parent cec84cf1bf
commit ec2b5b1810
7 changed files with 107 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Session.java
View File

@ -763,7 +763,7 @@ public abstract class HTTP2Session extends ContainerLifeCycle implements ISessio
int localCount = localStreamCount.get();
int maxCount = getMaxLocalStreams();
if (maxCount >= 0 && localCount >= maxCount)
throw new IllegalStateException("Max local stream count " + maxCount + " exceeded");
throw new IllegalStateException("Max local stream count " + maxCount + " exceeded" + System.lineSeparator() + dump());
if (localStreamCount.compareAndSet(localCount, localCount + 1))
break;
}

22
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Stream.java
View File

@ -510,6 +510,13 @@ public class HTTP2Stream extends IdleTimeout implements IStream, Callback, Dumpa
}
}
@Override
public void onClose()
{
super.onClose();
notifyClosed(this);
}
private void updateStreamCount(int deltaStream, int deltaClosing)
{
((HTTP2Session)session).updateStreamCount(isLocal(), deltaStream, deltaClosing);
@ -615,6 +622,21 @@ public class HTTP2Stream extends IdleTimeout implements IStream, Callback, Dumpa
}
}
private void notifyClosed(Stream stream)
{
Listener listener = this.listener;
if (listener == null)
return;
try
{
listener.onClosed(stream);
}
catch (Throwable x)
{
LOG.info("Failure while notifying listener " + listener, x);
}
}
@Override
public String dump()
{

9
jetty-http2/http2-common/src/main/java/org/eclipse/jetty/http2/api/Stream.java
View File

@ -234,6 +234,15 @@ public interface Stream
callback.succeeded();
}
/**
* <p>Callback method invoked after the stream has been closed.</p>
*
* @param stream the stream
*/
public default void onClosed(Stream stream)
{
}
/**
* <p>Empty implementation of {@link Listener}</p>
*/

5
jetty-http2/http2-http-client-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpChannelOverHTTP2.java
View File

@ -101,6 +101,11 @@ public class HttpChannelOverHTTP2 extends HttpChannel
connection.release(this);
}
void onStreamClosed(Stream stream)
{
connection.onStreamClosed(stream, this);
}
@Override
public void exchangeTerminated(HttpExchange exchange, Result result)
{

17
jetty-http2/http2-http-client-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpConnectionOverHTTP2.java
View File

@ -36,11 +36,16 @@ import org.eclipse.jetty.client.SendFailure;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.http2.ErrorCode;
import org.eclipse.jetty.http2.api.Session;
import org.eclipse.jetty.http2.api.Stream;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.Sweeper;
public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.Sweepable
{
private static final Logger LOG = Log.getLogger(HttpConnection.class);
private final Set<HttpChannel> activeChannels = ConcurrentHashMap.newKeySet();
private final Queue<HttpChannelOverHTTP2> idleChannels = new ConcurrentLinkedQueue<>();
private final AtomicBoolean closed = new AtomicBoolean();
@ -87,16 +92,16 @@ public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.S
protected void release(HttpChannelOverHTTP2 channel)
{
if (LOG.isDebugEnabled())
LOG.debug("Released {}", channel);
// Only non-push channels are released.
if (activeChannels.remove(channel))
{
channel.setStream(null);
// Recycle only non-failed channels.
if (channel.isFailed())
channel.destroy();
else
idleChannels.offer(channel);
getHttpDestination().release(this);
}
else
{
@ -104,6 +109,14 @@ public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.S
}
}
void onStreamClosed(Stream stream, HttpChannelOverHTTP2 channel)
{
if (LOG.isDebugEnabled())
LOG.debug("{} closed for {}", stream, channel);
channel.setStream(null);
getHttpDestination().release(this);
}
@Override
public boolean onIdleTimeout(long idleTimeout)
{

6
jetty-http2/http2-http-client-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpReceiverOverHTTP2.java
View File

@ -184,6 +184,12 @@ public class HttpReceiverOverHTTP2 extends HttpReceiver implements Stream.Listen
callback.succeeded();
}
@Override
public void onClosed(Stream stream)
{
getHttpChannel().onStreamClosed(stream);
}
private void notifyContent(HttpExchange exchange, DataFrame frame, Callback callback)
{
contentNotifier.offer(new DataInfo(exchange, frame, callback));

67
tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/HttpClientLoadTest.java
View File

@ -18,10 +18,6 @@
package org.eclipse.jetty.http.client;
import static org.eclipse.jetty.http.client.Transport.UNIX_SOCKET;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@ -30,11 +26,11 @@ import java.util.Locale;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@ -68,6 +64,9 @@ import org.hamcrest.Matchers;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ArgumentsSource;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTransportScenario>
{
private final Logger logger = Log.getLogger(HttpClientLoadTest.class);
@ -186,7 +185,7 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
// Choose a random method
HttpMethod method = random.nextBoolean() ? HttpMethod.GET : HttpMethod.POST;
boolean ssl = scenario.isTransportSecure();
boolean ssl = scenario.transport.isTlsBased();
// Choose randomly whether to close the connection on the client or on the server
boolean clientClose = false;
@ -196,13 +195,17 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
if (!ssl && random.nextInt(100) < 5)
serverClose = true;
int maxContentLength = 64 * 1024;
long clientTimeout = 0;
// if (!ssl && random.nextInt(100) < 5)
// clientTimeout = random.nextInt(500) + 500;
int maxContentLength = 1024 * 1024;
int contentLength = random.nextInt(maxContentLength) + 1;
test(scenario.getScheme(), host, method.asString(), clientClose, serverClose, contentLength, true, latch, failures);
test(scenario.getScheme(), host, method.asString(), clientClose, serverClose, clientTimeout, contentLength, true, latch, failures);
}
private void test(String scheme, String host, String method, boolean clientClose, boolean serverClose, int contentLength, final boolean checkContentLength, final CountDownLatch latch, final List<String> failures)
private void test(String scheme, String host, String method, boolean clientClose, boolean serverClose, long clientTimeout, int contentLength, final boolean checkContentLength, final CountDownLatch latch, final List<String> failures)
{
long requestId = requestCount.incrementAndGet();
Request request = scenario.client.newRequest(host, scenario.getNetworkConnectorLocalPortInt().orElse(0))
@ -215,6 +218,12 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
else if (serverClose)
request.header("X-Close", "true");
if (clientTimeout > 0)
{
request.header("X-Timeout", String.valueOf(clientTimeout));
request.idleTimeout(clientTimeout, TimeUnit.MILLISECONDS);
}
switch (method)
{
case "GET":
@ -254,12 +263,18 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
{
if (result.isFailed())
{
result.getFailure().printStackTrace();
failures.add("Result failed " + result);
Throwable failure = result.getFailure();
if (!(clientTimeout > 0 && failure instanceof TimeoutException))
{
failure.printStackTrace();
failures.add("Result failed " + result);
}
}
else
{
if (checkContentLength && contentLength.get() != 0)
failures.add("Content length mismatch " + contentLength);
}
if (checkContentLength && contentLength.get() != 0)
failures.add("Content length mismatch " + contentLength);
requestLatch.countDown();
latch.countDown();
@ -288,8 +303,14 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
private class LoadHandler extends AbstractHandler
{
@Override
public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException
{
baseRequest.setHandled(true);
String timeout = request.getHeader("X-Timeout");
if (timeout != null)
sleep(2 * Integer.parseInt(timeout));
String method = request.getMethod().toUpperCase(Locale.ENGLISH);
switch (method)
{
@ -313,8 +334,17 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
if (Boolean.parseBoolean(request.getHeader("X-Close")))
response.setHeader("Connection", "close");
}
baseRequest.setHandled(true);
private void sleep(long time)
{
try
{
Thread.sleep(time);
}
catch (InterruptedException ignored)
{
}
}
}
@ -329,8 +359,9 @@ public class HttpClientLoadTest extends AbstractTest<HttpClientLoadTest.LoadTran
}
@Override
public Connector newServerConnector( Server server) throws Exception {
if (transport == UNIX_SOCKET)
public Connector newServerConnector( Server server)
{
if (transport == Transport.UNIX_SOCKET)
{
UnixSocketConnector
unixSocketConnector = new UnixSocketConnector( server, provideServerConnectionFactory( transport ));