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

461499 - ConnectionPool may leak connections. 

Now using SpinLock to guard concurrent access to the connection
queues (idle and active) so that operations on them are atomic.
This commit is contained in:
Simone Bordet 2015-03-05 10:08:34 +01:00
parent 2169ea6494
commit 1a5346ec4f
6 changed files with 438 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
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;
@ -33,23 +35,25 @@ 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 Promise<Connection> connectionPromise;
private final Promise<Connection> requester;
private final BlockingDeque<Connection> idleConnections;
private final BlockingQueue<Connection> activeConnections;
public ConnectionPool(Destination destination, int maxConnections, Promise<Connection> connectionPromise)
public ConnectionPool(Destination destination, int maxConnections, Promise<Connection> requester)
{
this.destination = destination;
this.maxConnections = maxConnections;
this.connectionPromise = connectionPromise;
this.requester = requester;
this.idleConnections = new LinkedBlockingDeque<>(maxConnections);
this.activeConnections = new BlockingArrayQueue<>(maxConnections);
}
@ -71,7 +75,7 @@ public class ConnectionPool implements Closeable, Dumpable
public Connection acquire()
{
Connection connection = acquireIdleConnection();
Connection connection = activateIdle();
if (connection == null)
connection = tryCreate();
return connection;
@ -89,7 +93,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))
@ -104,10 +108,16 @@ public class ConnectionPool implements Closeable, Dumpable
{
if (LOG.isDebugEnabled())
LOG.debug("Connection {}/{} creation succeeded {}", next, maxConnections, connection);
if (activate(connection))
connectionPromise.succeeded(connection);
else
connectionPromise.failed(new IllegalStateException("Active connection overflow"));
boolean idle;
try (SpinLock.Lock lock = ConnectionPool.this.lock.lock())
{
// Use "cold" new connections as last.
idle = idleConnections.offerLast(connection);
}
idle(connection, idle);
requester.succeeded(connection);
}
@Override
@ -115,40 +125,44 @@ public class ConnectionPool implements Closeable, Dumpable
{
if (LOG.isDebugEnabled())
LOG.debug("Connection " + next + "/" + maxConnections + " creation failed", x);
connectionCount.decrementAndGet();
connectionPromise.failed(x);
requester.failed(x);
}
});
// Try again the idle connections
return acquireIdleConnection();
return activateIdle();
}
}
}
private Connection acquireIdleConnection()
private Connection activateIdle()
{
Connection connection = idleConnections.pollFirst();
if (connection == null)
return null;
return activate(connection) ? connection : null;
}
boolean acquired;
Connection connection;
try (SpinLock.Lock lock = this.lock.lock())
{
connection = idleConnections.pollFirst();
if (connection == null)
return null;
acquired = activeConnections.offer(connection);
}
private boolean activate(Connection connection)
{
if (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;
}
}
@ -158,24 +172,33 @@ public class ConnectionPool implements Closeable, Dumpable
public boolean release(Connection connection)
{
released(connection);
if (activeConnections.remove(connection))
boolean idle;
try (SpinLock.Lock lock = this.lock.lock())
{
// Make sure we use "hot" connections first
if (idleConnections.offerFirst(connection))
{
if (LOG.isDebugEnabled())
LOG.debug("Connection idle {}", connection);
return true;
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("Connection idle overflow {}", connection);
connection.close();
}
if (!activeConnections.remove(connection))
return false;
// Make sure we use "hot" connections first.
idle = idleConnections.offerFirst(connection);
}
released(connection);
return idle(connection, idle);
}
private boolean idle(Connection connection, boolean idle)
{
if (idle)
{
if (LOG.isDebugEnabled())
LOG.debug("Connection idle {}", connection);
return true;
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("Connection idle overflow {}", connection);
connection.close();
return false;
}
return false;
}
protected void released(Connection connection)
@ -184,9 +207,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)
@ -200,12 +228,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()
@ -215,16 +249,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
@ -236,18 +277,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);
}
}

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

@ -50,7 +50,7 @@ public abstract class PoolingHttpDestination<C extends Connection> extends HttpD
@SuppressWarnings("unchecked")
public void succeeded(Connection connection)
{
process((C)connection, true);
send(true);
}
@Override
@ -66,11 +66,19 @@ public abstract class PoolingHttpDestination<C extends Connection> extends HttpD
});
}
@Override
protected void send()
{
send(false);
}
private void send(boolean dispatch)
{
if (getHttpExchanges().isEmpty())
return;
C connection = acquire();
if (connection != null)
process(connection, false);
process(connection, dispatch);
}
@SuppressWarnings("unchecked")
@ -167,7 +175,6 @@ public abstract class PoolingHttpDestination<C extends Connection> extends HttpD
{
if (LOG.isDebugEnabled())
LOG.debug("{} is stopped", client);
close(connection);
connection.close();
}
}
@ -176,6 +183,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

34
jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java
View File

@ -19,6 +19,7 @@
package org.eclipse.jetty.client;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpCookie;
import java.net.URI;
@ -323,6 +324,7 @@ public class HttpClientTest extends AbstractHttpClientServerTest
public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
consume(request.getInputStream());
String value = request.getParameter(paramName);
if (paramValue.equals(value))
{
@ -347,9 +349,17 @@ public class HttpClientTest extends AbstractHttpClientServerTest
@Test
public void test_POST_WithContent_NotifiesRequestContentListener() throws Exception
{
final byte[] content = {0, 1, 2, 3};
start(new EmptyServerHandler());
start(new AbstractHandler()
{
@Override
public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
consume(request.getInputStream());
}
});
final byte[] content = {0, 1, 2, 3};
ContentResponse response = client.POST(scheme + "://localhost:" + connector.getLocalPort())
.onRequestContent(new Request.ContentListener()
{
@ -373,7 +383,15 @@ public class HttpClientTest extends AbstractHttpClientServerTest
@Test
public void test_POST_WithContent_TracksProgress() throws Exception
{
start(new EmptyServerHandler());
start(new AbstractHandler()
{
@Override
public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
consume(request.getInputStream());
}
});
final AtomicInteger progress = new AtomicInteger();
ContentResponse response = client.POST(scheme + "://localhost:" + connector.getLocalPort())
@ -1346,6 +1364,7 @@ public class HttpClientTest extends AbstractHttpClientServerTest
int count = requests.incrementAndGet();
if (count == maxRetries)
baseRequest.setHandled(true);
consume(request.getInputStream());
}
});
@ -1411,6 +1430,15 @@ public class HttpClientTest extends AbstractHttpClientServerTest
Assert.assertTrue(completeLatch.await(5, TimeUnit.SECONDS));
}
private void consume(InputStream input) throws IOException
{
while (true)
{
if (input.read() < 0)
break;
}
}
public static abstract class RetryListener implements Response.CompleteListener
{
private final HttpClient client;

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

@ -0,0 +1,284 @@
//
// ========================================================================
// 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.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.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)
{
return new HttpConnectionOverHTTP(endPoint, destination)
{
@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());
}
}

9
jetty-client/src/test/java/org/eclipse/jetty/client/HttpRequestAbortTest.java
View File

@ -87,16 +87,11 @@ public class HttpRequestAbortTest extends AbstractHttpClientServerTest
Assert.assertFalse(begin.get());
}
// The request send triggered a connection creation
// that is not awaited before failing the exchange.
Thread.sleep(1000);
// However, the connection has not been used, so it's a good one.
HttpDestinationOverHTTP destination = (HttpDestinationOverHTTP)client.getDestination(scheme, "localhost", connector.getLocalPort());
ConnectionPool connectionPool = destination.getConnectionPool();
Assert.assertEquals(1, connectionPool.getConnectionCount());
Assert.assertEquals(0, connectionPool.getConnectionCount());
Assert.assertEquals(0, connectionPool.getActiveConnections().size());
Assert.assertEquals(1, connectionPool.getIdleConnections().size());
Assert.assertEquals(0, connectionPool.getIdleConnections().size());
}
@Test