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

484210 - HttpClient over HTTP/2 should honor maxConcurrentStreams. 

Fixed by sending queued requests in a loop up to maxConcurrentStreams.
Also updating the maxConcurrentStreams value when received from the
server.
This commit is contained in:
Simone Bordet 2015-12-11 18:00:48 +01:00
parent e674d3ec5e
commit 8d28be5786
5 changed files with 316 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
jetty-client/src/main/java/org/eclipse/jetty/client/MultiplexHttpDestination.java
View File

@ -18,6 +18,7 @@
package org.eclipse.jetty.client;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.client.api.Connection;
@ -27,6 +28,8 @@ import org.eclipse.jetty.util.Promise;
public abstract class MultiplexHttpDestination<C extends Connection> extends HttpDestination implements Promise<Connection>
{
private final AtomicReference<ConnectState> connect = new AtomicReference<>(ConnectState.DISCONNECTED);
private final AtomicInteger requestsPerConnection = new AtomicInteger();
private int maxRequestsPerConnection = 1024;
private C connection;
protected MultiplexHttpDestination(HttpClient client, Origin origin)
@ -34,6 +37,16 @@ public abstract class MultiplexHttpDestination<C extends Connection> extends Htt
super(client, origin);
}
public int getMaxRequestsPerConnection()
{
return maxRequestsPerConnection;
}
public void setMaxRequestsPerConnection(int maxRequestsPerConnection)
{
this.maxRequestsPerConnection = maxRequestsPerConnection;
}
@Override
public void send()
{
@ -56,8 +69,7 @@ public abstract class MultiplexHttpDestination<C extends Connection> extends Htt
}
case CONNECTED:
{
if (process(connection))
break;
process(connection);
return;
}
default:
@ -92,36 +104,51 @@ public abstract class MultiplexHttpDestination<C extends Connection> extends Htt
abort(x);
}
protected boolean process(final C connection)
protected void process(final C connection)
{
HttpClient client = getHttpClient();
final HttpExchange exchange = getHttpExchanges().poll();
if (LOG.isDebugEnabled())
LOG.debug("Processing {} on {}", exchange, connection);
if (exchange == null)
return false;
while (true)
{
int max = getMaxRequestsPerConnection();
int count = requestsPerConnection.get();
int next = count + 1;
if (next > max)
return;
final Request request = exchange.getRequest();
Throwable cause = request.getAbortCause();
if (cause != null)
{
if (LOG.isDebugEnabled())
LOG.debug("Aborted before processing {}: {}", exchange, cause);
// It may happen that the request is aborted before the exchange
// is created. Aborting the exchange a second time will result in
// a no-operation, so we just abort here to cover that edge case.
exchange.abort(cause);
if (requestsPerConnection.compareAndSet(count, next))
{
HttpExchange exchange = getHttpExchanges().poll();
if (LOG.isDebugEnabled())
LOG.debug("Processing {}/{} {} on {}", next, max, exchange, connection);
if (exchange == null)
{
requestsPerConnection.decrementAndGet();
return;
}
final Request request = exchange.getRequest();
Throwable cause = request.getAbortCause();
if (cause != null)
{
if (LOG.isDebugEnabled())
LOG.debug("Aborted before processing {}: {}", exchange, cause);
// It may happen that the request is aborted before the exchange
// is created. Aborting the exchange a second time will result in
// a no-operation, so we just abort here to cover that edge case.
exchange.abort(cause);
requestsPerConnection.decrementAndGet();
}
else
{
send(connection, exchange);
}
}
}
else
{
send(connection, exchange);
}
return true;
}
@Override
public void release(Connection connection)
{
requestsPerConnection.decrementAndGet();
send();
}

17
jetty-http2/http2-http-client-transport/pom.xml
View File

@ -61,6 +61,23 @@
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.toolchain</groupId>
<artifactId>jetty-test-helper</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.http2</groupId>
<artifactId>http2-server</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>

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

@ -33,6 +33,7 @@ import org.eclipse.jetty.http2.api.Session;
import org.eclipse.jetty.http2.client.HTTP2Client;
import org.eclipse.jetty.http2.client.HTTP2ClientConnectionFactory;
import org.eclipse.jetty.http2.frames.GoAwayFrame;
import org.eclipse.jetty.http2.frames.SettingsFrame;
import org.eclipse.jetty.io.ClientConnectionFactory;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.ssl.SslClientConnectionFactory;
@ -106,9 +107,9 @@ public class HttpClientTransportOverHTTP2 extends ContainerLifeCycle implements
{
client.setConnectTimeout(httpClient.getConnectTimeout());
final HttpDestination destination = (HttpDestination)context.get(HTTP_DESTINATION_CONTEXT_KEY);
HttpDestinationOverHTTP2 destination = (HttpDestinationOverHTTP2)context.get(HTTP_DESTINATION_CONTEXT_KEY);
@SuppressWarnings("unchecked")
final Promise<Connection> connectionPromise = (Promise<Connection>)context.get(HTTP_CONNECTION_PROMISE_CONTEXT_KEY);
Promise<Connection> connectionPromise = (Promise<Connection>)context.get(HTTP_CONNECTION_PROMISE_CONTEXT_KEY);
SessionListenerPromise listenerPromise = new SessionListenerPromise(destination, connectionPromise);
@ -136,11 +137,11 @@ public class HttpClientTransportOverHTTP2 extends ContainerLifeCycle implements
private class SessionListenerPromise extends Session.Listener.Adapter implements Promise<Session>
{
private final HttpDestination destination;
private final HttpDestinationOverHTTP2 destination;
private final Promise<Connection> promise;
private HttpConnectionOverHTTP2 connection;
public SessionListenerPromise(HttpDestination destination, Promise<Connection> promise)
public SessionListenerPromise(HttpDestinationOverHTTP2 destination, Promise<Connection> promise)
{
this.destination = destination;
this.promise = promise;
@ -159,6 +160,14 @@ public class HttpClientTransportOverHTTP2 extends ContainerLifeCycle implements
promise.failed(failure);
}
@Override
public void onSettings(Session session, SettingsFrame frame)
{
Map<Integer, Integer> settings = frame.getSettings();
if (settings.containsKey(SettingsFrame.MAX_CONCURRENT_STREAMS))
destination.setMaxRequestsPerConnection(settings.get(SettingsFrame.MAX_CONCURRENT_STREAMS));
}
@Override
public void onClose(Session session, GoAwayFrame frame)
{

70
jetty-http2/http2-http-client-transport/src/test/java/org/eclipse/jetty/http2/client/http/AbstractTest.java Normal file
View File

@ -0,0 +1,70 @@
//
// ========================================================================
// Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.http2.client.http;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.http2.client.HTTP2Client;
import org.eclipse.jetty.http2.server.HTTP2ServerConnectionFactory;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.toolchain.test.TestTracker;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.junit.After;
import org.junit.Rule;
public class AbstractTest
{
@Rule
public TestTracker tracker = new TestTracker();
protected Server server;
protected ServerConnector connector;
protected HttpClient client;
protected void start(int maxConcurrentStreams, Handler handler) throws Exception
{
QueuedThreadPool serverExecutor = new QueuedThreadPool();
serverExecutor.setName("server");
server = new Server(serverExecutor);
HTTP2ServerConnectionFactory http2 = new HTTP2ServerConnectionFactory(new HttpConfiguration());
http2.setMaxConcurrentStreams(maxConcurrentStreams);
connector = new ServerConnector(server, 1, 1, http2);
server.addConnector(connector);
server.setHandler(handler);
server.start();
client = new HttpClient(new HttpClientTransportOverHTTP2(new HTTP2Client()), null);
QueuedThreadPool clientExecutor = new QueuedThreadPool();
clientExecutor.setName("client");
client.setExecutor(clientExecutor);
client.start();
}
@After
public void dispose() throws Exception
{
if (client != null)
client.stop();
if (server != null)
server.stop();
}
}

165
jetty-http2/http2-http-client-transport/src/test/java/org/eclipse/jetty/http2/client/http/MaxConcurrentStreamsTest.java Normal file
View File

@ -0,0 +1,165 @@
//
// ========================================================================
// Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.http2.client.http;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.client.api.ContentResponse;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.handler.AbstractHandler;
import org.junit.Assert;
import org.junit.Test;
public class MaxConcurrentStreamsTest extends AbstractTest
{
@Test
public void testOneConcurrentStream() throws Exception
{
long sleep = 1000;
start(1, new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
// Sleep a bit to allow the second request to be queued.
sleep(sleep);
}
});
// Prime the connection so that the maxConcurrentStream setting arrives to the client.
client.newRequest("localhost", connector.getLocalPort()).path("/prime").send();
CountDownLatch latch = new CountDownLatch(2);
// First request is sent immediately.
client.newRequest("localhost", connector.getLocalPort())
.path("/first")
.send(result ->
{
if (result.isSucceeded())
latch.countDown();
});
// Second request is queued.
client.newRequest("localhost", connector.getLocalPort())
.path("/second")
.send(result ->
{
if (result.isSucceeded())
latch.countDown();
});
// When the first request returns, the second must be sent.
Assert.assertTrue(latch.await(5 * sleep, TimeUnit.MILLISECONDS));
}
@Test
public void testTwoConcurrentStreamsThirdWaits() throws Exception
{
int maxStreams = 2;
long sleep = 1000;
start(maxStreams, new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
sleep(sleep);
}
});
// Prime the connection so that the maxConcurrentStream setting arrives to the client.
client.newRequest("localhost", connector.getLocalPort()).path("/prime").send();
// Send requests up to the max allowed.
for (int i = 0; i < maxStreams; ++i)
{
client.newRequest("localhost", connector.getLocalPort())
.path("/" + i)
.send(null);
}
// Send the request in excess.
CountDownLatch latch = new CountDownLatch(1);
String path = "/excess";
client.newRequest("localhost", connector.getLocalPort())
.path(path)
.send(result ->
{
if (result.getResponse().getStatus() == HttpStatus.OK_200)
latch.countDown();
});
// The last exchange should remain in the queue.
HttpDestinationOverHTTP2 destination = (HttpDestinationOverHTTP2)client.getDestination("http", "localhost", connector.getLocalPort());
Assert.assertEquals(1, destination.getHttpExchanges().size());
Assert.assertEquals(path, destination.getHttpExchanges().peek().getRequest().getPath());
Assert.assertTrue(latch.await(5 * sleep, TimeUnit.MILLISECONDS));
}
@Test
public void testAbortedWhileQueued() throws Exception
{
long sleep = 1000;
start(1, new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
sleep(sleep);
}
});
// Prime the connection so that the maxConcurrentStream setting arrives to the client.
client.newRequest("localhost", connector.getLocalPort()).path("/prime").send();
// Send a request that is aborted while queued.
client.newRequest("localhost", connector.getLocalPort())
.path("/aborted")
.onRequestQueued(request -> request.abort(new Exception()))
.send(null);
// Must be able to send another request.
ContentResponse response = client.newRequest("localhost", connector.getLocalPort()).path("/check").send();
Assert.assertEquals(HttpStatus.OK_200, response.getStatus());
}
private void sleep(long time)
{
try
{
Thread.sleep(time);
}
catch (InterruptedException x)
{
throw new RuntimeException(x);
}
}
}