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

Merged branch 'jetty-9.3.x' into 'master'.

This commit is contained in:
Simone Bordet 2015-12-11 21:39:03 +01:00
parent 1bba40a549 8d28be5786
commit 829bdd3d7c
8 changed files with 355 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java
View File

@ -276,12 +276,18 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest
private void process()
{
Connection connection = connectionPool.acquire();
if (connection != null)
process(connection);
while (true)
{
Connection connection = connectionPool.acquire();
if (connection == null)
break;
boolean proceed = process(connection);
if (!proceed)
break;
}
}
public void process(final Connection connection)
public boolean process(final Connection connection)
{
HttpClient client = getHttpClient();
final HttpExchange exchange = getHttpExchanges().poll();
@ -291,13 +297,13 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest
{
if (!connectionPool.release(connection))
connection.close();
if (!client.isRunning())
{
if (LOG.isDebugEnabled())
LOG.debug("{} is stopping", client);
connection.close();
}
return false;
}
else
{
@ -316,6 +322,7 @@ public abstract class HttpDestination extends ContainerLifeCycle implements Dest
{
send(connection, exchange);
}
return getHttpExchanges().peek() != null;
}
}

34
jetty-client/src/main/java/org/eclipse/jetty/client/MultiplexConnectionPool.java
View File

@ -40,18 +40,18 @@ public class MultiplexConnectionPool extends AbstractConnectionPool
private static final Logger LOG = Log.getLogger(MultiplexConnectionPool.class);
private final ReentrantLock lock = new ReentrantLock();
private final int maxMultiplexed;
private final Deque<Holder> idleConnections;
private final Map<Connection, Holder> muxedConnections;
private final Map<Connection, Holder> busyConnections;
private int maxMultiplex;
public MultiplexConnectionPool(HttpDestination destination, int maxConnections, Callback requester, int maxMultiplexed)
public MultiplexConnectionPool(HttpDestination destination, int maxConnections, Callback requester, int maxMultiplex)
{
super(destination, maxConnections, requester);
this.maxMultiplexed = maxMultiplexed;
this.idleConnections = new ArrayDeque<>(maxConnections);
this.muxedConnections = new HashMap<>(maxConnections);
this.busyConnections = new HashMap<>(maxConnections);
this.maxMultiplex = maxMultiplex;
}
protected void lock()
@ -64,6 +64,32 @@ public class MultiplexConnectionPool extends AbstractConnectionPool
lock.unlock();
}
public int getMaxMultiplex()
{
lock();
try
{
return maxMultiplex;
}
finally
{
unlock();
}
}
public void setMaxMultiplex(int maxMultiplex)
{
lock();
try
{
this.maxMultiplex = maxMultiplex;
}
finally
{
unlock();
}
}
@Override
public boolean isActive(Connection connection)
{
@ -120,7 +146,7 @@ public class MultiplexConnectionPool extends AbstractConnectionPool
holder = muxedConnections.values().iterator().next();
}
if (holder.count < maxMultiplexed)
if (holder.count < maxMultiplex)
{
++holder.count;
break;

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

@ -30,4 +30,19 @@ public abstract class MultiplexHttpDestination extends HttpDestination
return new MultiplexConnectionPool(this, client.getMaxConnectionsPerDestination(), this,
client.getMaxRequestsQueuedPerDestination());
}
public int getMaxRequestsPerConnection()
{
ConnectionPool connectionPool = getConnectionPool();
if (connectionPool instanceof MultiplexConnectionPool)
return ((MultiplexConnectionPool)connectionPool).getMaxMultiplex();
return 1;
}
public void setMaxRequestsPerConnection(int maxRequestsPerConnection)
{
ConnectionPool connectionPool = getConnectionPool();
if (connectionPool instanceof MultiplexConnectionPool)
((MultiplexConnectionPool)connectionPool).setMaxMultiplex(maxRequestsPerConnection);
}
}

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)
{

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

@ -57,6 +57,7 @@ public class HttpConnectionOverHTTP2 extends HttpConnection
protected void release(HttpChannel channel)
{
channels.remove(channel);
getHttpDestination().release(this);
}
@Override

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();
}
}

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

@ -0,0 +1,197 @@
//
// ========================================================================
// 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);
}
});
client.setMaxConnectionsPerDestination(1);
// 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);
}
});
client.setMaxConnectionsPerDestination(1);
// 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);
}
});
client.setMaxConnectionsPerDestination(1);
// 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());
}
@Test
public void testMultipleRequestsQueuedOnConnect() throws Exception
{
int maxConcurrent = 10;
long sleep = 500;
start(maxConcurrent, new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
sleep(sleep);
}
});
client.setMaxConnectionsPerDestination(1);
// The first request will open the connection, the others will be queued.
CountDownLatch latch = new CountDownLatch(maxConcurrent);
for (int i = 0; i < maxConcurrent; ++i)
{
client.newRequest("localhost", connector.getLocalPort())
.path("/" + i)
.send(result -> latch.countDown());
}
// The requests should be processed in parallel, not sequentially.
Assert.assertTrue(latch.await(maxConcurrent * sleep / 2, TimeUnit.MILLISECONDS));
}
private void sleep(long time)
{
try
{
Thread.sleep(time);
}
catch (InterruptedException x)
{
throw new RuntimeException(x);
}
}
}