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.2.x' into 'master'.

This commit is contained in:
Simone Bordet 2015-03-05 18:19:27 +01:00
parent 48b6bec64c 3b8bf8d1e0
commit 3e0949ad2c
5 changed files with 386 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
jetty-client/src/main/java/org/eclipse/jetty/client/ConnectionPool.java
View File

@ -20,6 +20,8 @@ package org.eclipse.jetty.client;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
@ -34,12 +36,14 @@ import org.eclipse.jetty.util.component.ContainerLifeCycle;
import org.eclipse.jetty.util.component.Dumpable;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.SpinLock;
public class ConnectionPool implements Closeable, Dumpable
{
protected static final Logger LOG = Log.getLogger(ConnectionPool.class);
private final AtomicInteger connectionCount = new AtomicInteger();
private final SpinLock lock = new SpinLock();
private final Destination destination;
private final int maxConnections;
private final Callback requester;
@ -72,7 +76,7 @@ public class ConnectionPool implements Closeable, Dumpable
public Connection acquire()
{
Connection connection = acquireIdleConnection();
Connection connection = activateIdle();
if (connection == null)
connection = tryCreate();
return connection;
@ -90,7 +94,7 @@ public class ConnectionPool implements Closeable, Dumpable
if (LOG.isDebugEnabled())
LOG.debug("Max connections {}/{} reached", current, maxConnections);
// Try again the idle connections
return acquireIdleConnection();
return activateIdle();
}
if (connectionCount.compareAndSet(current, next))
@ -105,7 +109,9 @@ public class ConnectionPool implements Closeable, Dumpable
{
if (LOG.isDebugEnabled())
LOG.debug("Connection {}/{} creation succeeded {}", next, maxConnections, connection);
idle(connection, true);
idleCreated(connection);
requester.succeeded();
}
@ -114,40 +120,55 @@ public class ConnectionPool implements Closeable, Dumpable
{
if (LOG.isDebugEnabled())
LOG.debug("Connection " + next + "/" + maxConnections + " creation failed", x);
connectionCount.decrementAndGet();
requester.failed(x);
}
});
// Try again the idle connections
return acquireIdleConnection();
return activateIdle();
}
}
}
private Connection acquireIdleConnection()
protected void idleCreated(Connection connection)
{
Connection connection = idleConnections.pollFirst();
if (connection == null)
return null;
return activate(connection) ? connection : null;
boolean idle;
try (SpinLock.Lock lock = this.lock.lock())
{
// Use "cold" new connections as last.
idle = idleConnections.offerLast(connection);
}
idle(connection, idle);
}
private boolean activate(Connection connection)
private Connection activateIdle()
{
if (activeConnections.offer(connection))
boolean acquired;
Connection connection;
try (SpinLock.Lock lock = this.lock.lock())
{
connection = idleConnections.pollFirst();
if (connection == null)
return null;
acquired = activeConnections.offer(connection);
}
if (acquired)
{
if (LOG.isDebugEnabled())
LOG.debug("Connection active {}", connection);
acquired(connection);
return true;
return connection;
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("Connection active overflow {}", connection);
connection.close();
return false;
return null;
}
}
@ -157,17 +178,20 @@ public class ConnectionPool implements Closeable, Dumpable
public boolean release(Connection connection)
{
boolean idle;
try (SpinLock.Lock lock = this.lock.lock())
{
if (!activeConnections.remove(connection))
return false;
// Make sure we use "hot" connections first.
idle = idleConnections.offerFirst(connection);
}
released(connection);
if (activeConnections.remove(connection))
return idle(connection, false);
return false;
return idle(connection, idle);
}
protected boolean idle(Connection connection, boolean created)
protected boolean idle(Connection connection, boolean idle)
{
// Make sure we use "hot" connections first.
boolean idle = created ? idleConnections.offerLast(connection)
: idleConnections.offerFirst(connection);
if (idle)
{
if (LOG.isDebugEnabled())
@ -189,9 +213,14 @@ public class ConnectionPool implements Closeable, Dumpable
public boolean remove(Connection connection)
{
boolean activeRemoved = activeConnections.remove(connection);
boolean idleRemoved = idleConnections.remove(connection);
if (!idleRemoved)
boolean activeRemoved;
boolean idleRemoved;
try (SpinLock.Lock lock = this.lock.lock())
{
activeRemoved = activeConnections.remove(connection);
idleRemoved = idleConnections.remove(connection);
}
if (activeRemoved)
released(connection);
boolean removed = activeRemoved || idleRemoved;
if (removed)
@ -205,12 +234,18 @@ public class ConnectionPool implements Closeable, Dumpable
public boolean isActive(Connection connection)
{
return activeConnections.contains(connection);
try (SpinLock.Lock lock = this.lock.lock())
{
return activeConnections.contains(connection);
}
}
public boolean isIdle(Connection connection)
{
return idleConnections.contains(connection);
try (SpinLock.Lock lock = this.lock.lock())
{
return idleConnections.contains(connection);
}
}
public boolean isEmpty()
@ -220,16 +255,23 @@ public class ConnectionPool implements Closeable, Dumpable
public void close()
{
for (Connection connection : idleConnections)
List<Connection> idles = new ArrayList<>();
List<Connection> actives = new ArrayList<>();
try (SpinLock.Lock lock = this.lock.lock())
{
idles.addAll(idleConnections);
idleConnections.clear();
actives.addAll(activeConnections);
activeConnections.clear();
}
connectionCount.set(0);
for (Connection connection : idles)
connection.close();
idleConnections.clear();
// A bit drastic, but we cannot wait for all requests to complete
for (Connection connection : activeConnections)
for (Connection connection : actives)
connection.close();
activeConnections.clear();
connectionCount.set(0);
}
@Override
@ -241,18 +283,32 @@ public class ConnectionPool implements Closeable, Dumpable
@Override
public void dump(Appendable out, String indent) throws IOException
{
List<Connection> actives = new ArrayList<>();
List<Connection> idles = new ArrayList<>();
try (SpinLock.Lock lock = this.lock.lock())
{
actives.addAll(activeConnections);
idles.addAll(idleConnections);
}
ContainerLifeCycle.dumpObject(out, this);
ContainerLifeCycle.dump(out, indent, activeConnections, idleConnections);
ContainerLifeCycle.dump(out, indent, actives, idles);
}
@Override
public String toString()
{
int activeSize;
int idleSize;
try (SpinLock.Lock lock = this.lock.lock())
{
activeSize = activeConnections.size();
idleSize = idleConnections.size();
}
return String.format("%s[c=%d/%d,a=%d,i=%d]",
getClass().getSimpleName(),
connectionCount.get(),
maxConnections,
activeConnections.size(),
idleConnections.size());
activeSize,
idleSize);
}
}

2
jetty-client/src/main/java/org/eclipse/jetty/client/PoolingHttpDestination.java
View File

@ -146,7 +146,6 @@ public abstract class PoolingHttpDestination<C extends Connection> extends HttpD
{
if (LOG.isDebugEnabled())
LOG.debug("{} is stopped", client);
close(connection);
connection.close();
}
}
@ -155,6 +154,7 @@ public abstract class PoolingHttpDestination<C extends Connection> extends HttpD
public void close(Connection oldConnection)
{
super.close(oldConnection);
connectionPool.remove(oldConnection);
if (getHttpExchanges().isEmpty())

4
jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java
View File

@ -93,10 +93,11 @@ public class HttpChannelOverHTTP extends HttpChannel
public void exchangeTerminated(Result result)
{
super.exchangeTerminated(result);
Response response = result.getResponse();
HttpFields responseHeaders = response.getHeaders();
boolean close = result.isFailed() || receiver.isShutdown();
// Only check HTTP headers if there are no failures.
if (!close)
{
if (response.getVersion().compareTo(HttpVersion.HTTP_1_1) < 0)
@ -110,6 +111,7 @@ public class HttpChannelOverHTTP extends HttpChannel
close = responseHeaders.contains(HttpHeader.CONNECTION, HttpHeaderValue.CLOSE.asString());
}
}
if (close)
connection.close();
else

286
jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientUploadDuringServerShutdown.java Normal file
View File

@ -0,0 +1,286 @@
//
// ========================================================================
// 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.client;
import java.io.IOException;
import java.io.InputStream;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.client.api.Connection;
import org.eclipse.jetty.client.api.Response;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.http.HttpChannelOverHTTP;
import org.eclipse.jetty.client.http.HttpClientTransportOverHTTP;
import org.eclipse.jetty.client.http.HttpConnectionOverHTTP;
import org.eclipse.jetty.client.http.HttpDestinationOverHTTP;
import org.eclipse.jetty.client.util.BytesContentProvider;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.handler.AbstractHandler;
import org.eclipse.jetty.util.Promise;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.junit.Assert;
import org.junit.Test;
public class HttpClientUploadDuringServerShutdown
{
/**
* A server used in conjunction with {@link ClientSide}.
*/
public static class ServerSide
{
public static void main(String[] args) throws Exception
{
QueuedThreadPool serverThreads = new QueuedThreadPool();
serverThreads.setName("server");
Server server = new Server(serverThreads);
ServerConnector connector = new ServerConnector(server);
connector.setPort(8888);
server.addConnector(connector);
server.setHandler(new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
byte[] buffer = new byte[1024];
InputStream input = request.getInputStream();
while (true)
{
int read = input.read(buffer);
if (read < 0)
break;
long now = System.nanoTime();
long sleep = TimeUnit.MICROSECONDS.toNanos(1);
while (System.nanoTime() < now + sleep)
{
// Wait.
}
}
}
});
server.start();
}
}
/**
* An infinite loop of a client uploading content to the server.
* The server may be killed while this client is running, and the
* behavior should be that this client continues running, failing
* exchanges while the server is down, but succeeding them when
* the server is up and running.
*
* @see ServerSide
*/
public static class ClientSide
{
public static void main(String[] args) throws Exception
{
QueuedThreadPool clientThreads = new QueuedThreadPool();
clientThreads.setName("client");
HttpClient client = new HttpClient(new HttpClientTransportOverHTTP(2), null);
client.setMaxConnectionsPerDestination(2);
client.setIdleTimeout(10000);
client.setExecutor(clientThreads);
client.start();
Random random = new Random();
while (true)
{
int count = 1;
final CountDownLatch latch = new CountDownLatch(count);
for (int i = 0; i < count; ++i)
{
int length = 16 * 1024 * 1024 + random.nextInt(16 * 1024 * 1024);
client.newRequest("localhost", 8888)
.content(new BytesContentProvider(new byte[length]))
.send(new Response.CompleteListener()
{
@Override
public void onComplete(Result result)
{
latch.countDown();
}
});
long sleep = 1 + random.nextInt(10);
TimeUnit.MILLISECONDS.sleep(sleep);
}
latch.await();
}
}
}
@Test
public void testUploadDuringServerShutdown() throws Exception
{
final Logger LOG = Log.getLogger(getClass());
final AtomicReference<EndPoint> endPointRef = new AtomicReference<>();
final CountDownLatch serverLatch = new CountDownLatch(1);
QueuedThreadPool serverThreads = new QueuedThreadPool();
serverThreads.setName("server");
Server server = new Server(serverThreads);
ServerConnector connector = new ServerConnector(server);
server.addConnector(connector);
server.setHandler(new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
endPointRef.set(baseRequest.getHttpChannel().getEndPoint());
serverLatch.countDown();
}
});
server.start();
final AtomicBoolean afterSetup = new AtomicBoolean();
final CountDownLatch sendLatch = new CountDownLatch(1);
final CountDownLatch beginLatch = new CountDownLatch(1);
final CountDownLatch associateLatch = new CountDownLatch(1);
QueuedThreadPool clientThreads = new QueuedThreadPool();
clientThreads.setName("client");
HttpClient client = new HttpClient(new HttpClientTransportOverHTTP(1)
{
@Override
protected HttpConnectionOverHTTP newHttpConnection(EndPoint endPoint, HttpDestination destination, Promise<Connection> promise)
{
return new HttpConnectionOverHTTP(endPoint, destination, promise)
{
@Override
protected HttpChannelOverHTTP newHttpChannel()
{
return new HttpChannelOverHTTP(this)
{
@Override
public void send()
{
if (afterSetup.get())
{
associateLatch.countDown();
LOG.info("Sending...");
}
super.send();
}
};
}
@Override
protected void close(Throwable failure)
{
try
{
sendLatch.countDown();
beginLatch.await(5, TimeUnit.SECONDS);
super.close(failure);
}
catch (InterruptedException x)
{
x.printStackTrace();
}
}
@Override
protected boolean abort(Throwable failure)
{
try
{
associateLatch.await(5, TimeUnit.SECONDS);
LOG.info("Aborting... {}", getHttpChannel());
return super.abort(failure);
}
catch (InterruptedException x)
{
x.printStackTrace();
return false;
}
}
};
}
}, null);
client.setIdleTimeout(10000);
client.setExecutor(clientThreads);
client.start();
// Create one connection.
client.newRequest("localhost", connector.getLocalPort()).send();
Assert.assertTrue(serverLatch.await(5, TimeUnit.SECONDS));
afterSetup.set(true);
Thread.sleep(1000);
// Close the connection, so that the receiver is woken
// up and will call HttpConnectionOverHTTP.close().
EndPoint endPoint = endPointRef.get();
endPoint.close();
// Wait for close() so that the connection that
// is being closed is used to send the request.
Assert.assertTrue(sendLatch.await(5, TimeUnit.SECONDS));
final CountDownLatch completeLatch = new CountDownLatch(1);
client.newRequest("localhost", connector.getLocalPort())
.timeout(10, TimeUnit.SECONDS)
.onRequestBegin(new org.eclipse.jetty.client.api.Request.BeginListener()
{
@Override
public void onBegin(org.eclipse.jetty.client.api.Request request)
{
try
{
beginLatch.countDown();
completeLatch.await(5, TimeUnit.SECONDS);
}
catch (InterruptedException x)
{
x.printStackTrace();
}
}
})
.send(new Response.CompleteListener()
{
@Override
public void onComplete(Result result)
{
LOG.info("Completed");
completeLatch.countDown();
}
});
Assert.assertTrue(completeLatch.await(5, TimeUnit.SECONDS));
HttpDestinationOverHTTP destination = (HttpDestinationOverHTTP)client.getDestination("http", "localhost", connector.getLocalPort());
ConnectionPool pool = destination.getConnectionPool();
Assert.assertEquals(0, pool.getConnectionCount());
Assert.assertEquals(0, pool.getIdleConnections().size());
Assert.assertEquals(0, pool.getActiveConnections().size());
}
}

13
jetty-client/src/test/java/org/eclipse/jetty/client/http/HttpDestinationOverHTTPTest.java
View File

@ -101,18 +101,17 @@ public class HttpDestinationOverHTTPTest extends AbstractHttpClientServerTest
return new ConnectionPool(this, client.getMaxConnectionsPerDestination(), this)
{
@Override
protected boolean idle(Connection connection, boolean created)
protected void idleCreated(Connection connection)
{
try
{
idleLatch.countDown();
latch.await(5, TimeUnit.SECONDS);
return super.idle(connection, created);
super.idleCreated(connection);
}
catch (InterruptedException x)
{
x.printStackTrace();
return false;
}
}
};
@ -120,14 +119,14 @@ public class HttpDestinationOverHTTPTest extends AbstractHttpClientServerTest
};
Connection connection1 = destination.acquire();
// Make sure we entered idle().
// Make sure we entered idleCreated().
Assert.assertTrue(idleLatch.await(5, TimeUnit.SECONDS));
// There are no available existing connections, so
// acquire() returns null because we delayed idle() above.
// There are no available existing connections, so acquire()
// returns null because we delayed idleCreated() above
Assert.assertNull(connection1);
// Second attempt also returns null because we delayed idle() above.
// Second attempt also returns null because we delayed idleCreated() above.
Connection connection2 = destination.acquire();
Assert.assertNull(connection2);