Apache
/
httpcomponents-client
mirror of https://github.com/apache/httpcomponents-client.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

HTTPCLIENT-948: redesign of idle connection handling 

git-svn-id: https://svn.apache.org/repos/asf/httpcomponents/httpclient/trunk@954659 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Oleg Kalnichevski 2010-06-14 22:00:56 +00:00
parent 0f8ca31e92
commit a2ad042193
6 changed files with 191 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
httpclient/src/main/java/org/apache/http/impl/conn/IdleConnectionHandler.java
View File

@ -31,8 +31,6 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import org.apache.http.annotation.NotThreadSafe;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.HttpConnection;
@ -46,8 +44,10 @@ import org.apache.http.HttpConnection;
* @see org.apache.http.conn.ClientConnectionManager#closeIdleConnections
*
* @since 4.0
*
* @deprecated no longer used
*/
@NotThreadSafe
@Deprecated
public class IdleConnectionHandler {
private final Log log = LogFactory.getLog(getClass());

28
httpclient/src/main/java/org/apache/http/impl/conn/tsccm/AbstractConnPool.java
View File

@ -37,7 +37,6 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.http.annotation.GuardedBy;
import org.apache.http.annotation.ThreadSafe;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -56,8 +55,7 @@ import org.apache.http.impl.conn.IdleConnectionHandler;
* @since 4.0
*/
@ThreadSafe
@SuppressWarnings("deprecation")
@Deprecated
public abstract class AbstractConnPool implements RefQueueHandler {
private final Log log;
@ -67,17 +65,10 @@ public abstract class AbstractConnPool implements RefQueueHandler {
*/
protected final Lock poolLock;
/**
* References to issued connections.
* Must hold poolLock when accessing.
*/
/** References to issued connections */
@GuardedBy("poolLock")
protected Set<BasicPoolEntry> leasedConnections;
/** The handler for idle connections. Must hold poolLock when accessing. */
@GuardedBy("poolLock")
protected IdleConnectionHandler idleConnHandler;
/** The current total number of connections. */
@GuardedBy("poolLock")
protected int numConnections;
@ -85,12 +76,12 @@ public abstract class AbstractConnPool implements RefQueueHandler {
/** Indicates whether this pool is shut down. */
protected volatile boolean isShutDown;
@Deprecated
protected Set<BasicPoolEntryRef> issuedConnections;
@Deprecated
protected ReferenceQueue<Object> refQueue;
protected IdleConnectionHandler idleConnHandler;
/**
* Creates a new connection pool.
*/
@ -102,15 +93,10 @@ public abstract class AbstractConnPool implements RefQueueHandler {
this.poolLock = new ReentrantLock();
}
/**
* @deprecated do not sue
*/
@Deprecated
public void enableConnectionGC()
throws IllegalStateException {
}
/**
* Obtains a pool entry with a connection within the given timeout.
*
@ -158,11 +144,9 @@ public abstract class AbstractConnPool implements RefQueueHandler {
public abstract void freeEntry(BasicPoolEntry entry, boolean reusable, long validDuration, TimeUnit timeUnit)
;
@Deprecated
public void handleReference(Reference<?> ref) {
}
@Deprecated
protected abstract void handleLostEntry(HttpRoute route);
/**
@ -200,9 +184,7 @@ public abstract class AbstractConnPool implements RefQueueHandler {
/**
* Deletes all entries for closed connections.
*/
public abstract void deleteClosedConnections()
;
public abstract void deleteClosedConnections();
/**
* Shuts down this pool and all associated resources.

48
httpclient/src/main/java/org/apache/http/impl/conn/tsccm/BasicPoolEntry.java
View File

@ -27,6 +27,7 @@
package org.apache.http.impl.conn.tsccm;
import java.lang.ref.ReferenceQueue;
import java.util.concurrent.TimeUnit;
import org.apache.http.annotation.NotThreadSafe;
import org.apache.http.conn.OperatedClientConnection;
@ -42,6 +43,11 @@ import org.apache.http.impl.conn.AbstractPoolEntry;
@NotThreadSafe
public class BasicPoolEntry extends AbstractPoolEntry {
private final long created;
private long updated;
private long expiry;
/**
* @deprecated do not use
*/
@ -53,6 +59,7 @@ public class BasicPoolEntry extends AbstractPoolEntry {
if (route == null) {
throw new IllegalArgumentException("HTTP route may not be null");
}
this.created = System.currentTimeMillis();
}
/**
@ -67,6 +74,7 @@ public class BasicPoolEntry extends AbstractPoolEntry {
if (route == null) {
throw new IllegalArgumentException("HTTP route may not be null");
}
this.created = System.currentTimeMillis();
}
protected final OperatedClientConnection getConnection() {
@ -87,6 +95,46 @@ public class BasicPoolEntry extends AbstractPoolEntry {
super.shutdownEntry();
}
/**
* @since 4.1
*/
public long getCreated() {
return this.created;
}
/**
* @since 4.1
*/
public long getUpdated() {
return this.updated;
}
/**
* @since 4.1
*/
public long getExpiry() {
return this.expiry;
}
/**
* @since 4.1
*/
public void updateExpiry(long time, TimeUnit timeunit) {
this.updated = System.currentTimeMillis();
if (time > 0) {
this.expiry = this.updated + timeunit.toMillis(time);
} else {
this.expiry = Long.MAX_VALUE;
}
}
/**
* @since 4.1
*/
public boolean isExpired(long now) {
return now >= this.expiry;
}
}

177
httpclient/src/main/java/org/apache/http/impl/conn/tsccm/ConnPoolByRoute.java
View File

@ -26,27 +26,29 @@
package org.apache.http.impl.conn.tsccm;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Queue;
import java.util.LinkedList;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.annotation.GuardedBy;
import org.apache.http.annotation.ThreadSafe;
import org.apache.http.conn.routing.HttpRoute;
import org.apache.http.conn.ClientConnectionOperator;
import org.apache.http.conn.ConnectionPoolTimeoutException;
import org.apache.http.conn.OperatedClientConnection;
import org.apache.http.conn.params.ConnPerRoute;
import org.apache.http.conn.params.ConnManagerParams;
import org.apache.http.params.HttpParams;
/**
* A connection pool that maintains connections by route.
* This class is derived from <code>MultiThreadedHttpConnectionManager</code>
@ -63,33 +65,37 @@ import org.apache.http.params.HttpParams;
* @since 4.0
*/
@ThreadSafe
public class ConnPoolByRoute extends AbstractConnPool {
@SuppressWarnings("deprecation")
public class ConnPoolByRoute extends AbstractConnPool { //TODO: remove dependency on AbstractConnPool
private final Log log = LogFactory.getLog(getClass());
private final Lock poolLock;
/** Connection operator for this pool */
protected final ClientConnectionOperator operator;
/** Connections per route lookup */
protected final ConnPerRoute connPerRoute;
/** References to issued connections */
protected final Set<BasicPoolEntry> leasedConnections;
/** The list of free connections */
protected final Queue<BasicPoolEntry> freeConnections;
/** The list of WaitingThreads waiting for a connection */
protected final Queue<WaitingThread> waitingThreads;
/**
* A map of route-specific pools.
* Keys are of class {@link HttpRoute},
* values of class {@link RouteSpecificPool}.
*/
@GuardedBy("poolLock")
/** Map of route-specific pools */
protected final Map<HttpRoute, RouteSpecificPool> routeToPool;
@GuardedBy("poolLock")
protected volatile boolean shutdown;
protected volatile int maxTotalConnections;
protected volatile int numConnections;
/**
* Creates a new connection pool, managed by route.
*
@ -106,6 +112,8 @@ public class ConnPoolByRoute extends AbstractConnPool {
if (connPerRoute == null) {
throw new IllegalArgumentException("Connections per route may not be null");
}
this.poolLock = super.poolLock;
this.leasedConnections = super.leasedConnections;
this.operator = operator;
this.connPerRoute = connPerRoute;
this.maxTotalConnections = maxTotalConnections;
@ -114,6 +122,10 @@ public class ConnPoolByRoute extends AbstractConnPool {
this.routeToPool = createRouteToPoolMap();
}
protected Lock getLock() {
return this.poolLock;
}
/**
* Creates a new connection pool, managed by route.
*
@ -184,6 +196,16 @@ public class ConnPoolByRoute extends AbstractConnPool {
return new WaitingThread(cond, rospl);
}
private void closeConnection(final BasicPoolEntry entry) {
OperatedClientConnection conn = entry.getConnection();
if (conn != null) {
try {
conn.close();
} catch (IOException ex) {
log.debug("I/O error closing connection", ex);
}
}
}
/**
* Get a route-specific pool of available connections.
@ -214,8 +236,6 @@ public class ConnPoolByRoute extends AbstractConnPool {
return rospl;
}
//@@@ consider alternatives for gathering statistics
public int getConnectionsInPool(HttpRoute route) {
poolLock.lock();
@ -229,6 +249,15 @@ public class ConnPoolByRoute extends AbstractConnPool {
}
}
public int getConnectionsInPool() {
poolLock.lock();
try {
return numConnections;
} finally {
poolLock.unlock();
}
}
@Override
public PoolEntryRequest requestPoolEntry(
final HttpRoute route,
@ -296,9 +325,8 @@ public class ConnPoolByRoute extends AbstractConnPool {
while (entry == null) {
if (isShutDown) {
throw new IllegalStateException
("Connection pool shut down.");
if (shutdown) {
throw new IllegalStateException("Connection pool shut down");
}
if (log.isDebugEnabled()) {
@ -378,13 +406,10 @@ public class ConnPoolByRoute extends AbstractConnPool {
} finally {
poolLock.unlock();
}
return entry;
} // getEntry
}
// non-javadoc, see base class AbstractConnPool
@Override
public void freeEntry(BasicPoolEntry entry, boolean reusable, long validDuration, TimeUnit timeUnit) {
@ -396,10 +421,10 @@ public class ConnPoolByRoute extends AbstractConnPool {
poolLock.lock();
try {
if (isShutDown) {
if (shutdown) {
// the pool is shut down, release the
// connection's resources and get out of here
closeConnection(entry.getConnection());
closeConnection(entry);
return;
}
@ -415,8 +440,8 @@ public class ConnPoolByRoute extends AbstractConnPool {
"; keep alive for " + validDuration + " " + timeUnit.toString());
}
rospl.freeEntry(entry);
entry.updateExpiry(validDuration, timeUnit);
freeConnections.add(entry);
idleConnHandler.add(entry.getConnection(), validDuration, timeUnit);
} else {
rospl.dropEntry();
numConnections--;
@ -457,14 +482,13 @@ public class ConnPoolByRoute extends AbstractConnPool {
}
freeConnections.remove(entry);
boolean valid = idleConnHandler.remove(entry.getConnection());
if(!valid) {
if (entry.isExpired(System.currentTimeMillis())) {
// If the free entry isn't valid anymore, get rid of it
// and loop to find another one that might be valid.
if(log.isDebugEnabled())
log.debug("Closing expired free connection"
+ " [" + rospl.getRoute() + "][" + state + "]");
closeConnection(entry.getConnection());
closeConnection(entry);
// We use dropEntry instead of deleteEntry because the entry
// is no longer "free" (we just allocated it), and deleteEntry
// can only be used to delete free entries.
@ -547,7 +571,7 @@ public class ConnPoolByRoute extends AbstractConnPool {
poolLock.lock();
try {
closeConnection(entry.getConnection());
closeConnection(entry);
RouteSpecificPool rospl = getRoutePool(route, true);
rospl.deleteEntry(entry);
@ -556,8 +580,6 @@ public class ConnPoolByRoute extends AbstractConnPool {
routeToPool.remove(route);
}
idleConnHandler.remove(entry.getConnection());// not idle, but dead
} finally {
poolLock.unlock();
}
@ -580,7 +602,7 @@ public class ConnPoolByRoute extends AbstractConnPool {
if (entry != null) {
deleteEntry(entry);
} else if (log.isDebugEnabled()) {
log.debug("No free connection to delete.");
log.debug("No free connection to delete");
}
} finally {
@ -588,8 +610,6 @@ public class ConnPoolByRoute extends AbstractConnPool {
}
}
// non-javadoc, see base class AbstractConnPool
@Override
protected void handleLostEntry(HttpRoute route) {
@ -610,7 +630,6 @@ public class ConnPoolByRoute extends AbstractConnPool {
}
}
/**
* Notifies a waiting thread that a connection is available.
* This will wake a thread waiting in the specific route pool,
@ -672,29 +691,96 @@ public class ConnPoolByRoute extends AbstractConnPool {
poolLock.unlock();
}
}
// non-javadoc, see base class AbstractConnPool
/**
* Closes idle connections.
*
* @param idletime the time the connections should have been idle
* in order to be closed now
* @param tunit the unit for the <code>idletime</code>
*/
@Override
public void shutdown() {
public void closeIdleConnections(long idletime, TimeUnit tunit) {
if (tunit == null) {
throw new IllegalArgumentException("Time unit must not be null.");
}
if (idletime < 0) {
idletime = 0;
}
if (log.isDebugEnabled()) {
log.debug("Closing connections idle longer than " + idletime + " " + tunit);
}
// the latest time for which connections will be closed
long deadline = System.currentTimeMillis() - tunit.toMillis(idletime);
poolLock.lock();
try {
Iterator<BasicPoolEntry> iter = freeConnections.iterator();
while (iter.hasNext()) {
BasicPoolEntry entry = iter.next();
if (entry.getUpdated() <= deadline) {
if (log.isDebugEnabled()) {
log.debug("Closing connection last used @ " + new Date(entry.getUpdated()));
}
iter.remove();
deleteEntry(entry);
}
}
} finally {
poolLock.unlock();
}
}
super.shutdown();
@Override
public void closeExpiredConnections() {
log.debug("Closing expired connections");
long now = System.currentTimeMillis();
poolLock.lock();
try {
Iterator<BasicPoolEntry> iter = freeConnections.iterator();
while (iter.hasNext()) {
BasicPoolEntry entry = iter.next();
if (entry.isExpired(now)) {
if (log.isDebugEnabled()) {
log.debug("Closing connection expired @ " + new Date(entry.getExpiry()));
}
iter.remove();
deleteEntry(entry);
}
}
} finally {
poolLock.unlock();
}
}
@Override
public void shutdown() {
poolLock.lock();
try {
if (shutdown) {
return;
}
shutdown = true;
// close all connections that are issued to an application
Iterator<BasicPoolEntry> iter1 = leasedConnections.iterator();
while (iter1.hasNext()) {
BasicPoolEntry entry = iter1.next();
iter1.remove();
closeConnection(entry);
}
// close all free connections
//@@@ move this to base class?
Iterator<BasicPoolEntry> ibpe = freeConnections.iterator();
while (ibpe.hasNext()) {
BasicPoolEntry entry = ibpe.next();
ibpe.remove();
Iterator<BasicPoolEntry> iter2 = freeConnections.iterator();
while (iter2.hasNext()) {
BasicPoolEntry entry = iter2.next();
iter2.remove();
if (log.isDebugEnabled()) {
log.debug("Closing connection"
+ " [" + entry.getPlannedRoute() + "][" + entry.getState() + "]");
}
closeConnection(entry.getConnection());
closeConnection(entry);
}
// wake up all waiting threads
@ -732,6 +818,5 @@ public class ConnPoolByRoute extends AbstractConnPool {
return maxTotalConnections;
}
} // class ConnPoolByRoute
}

23
httpclient/src/main/java/org/apache/http/impl/conn/tsccm/ThreadSafeClientConnManager.java
View File

@ -284,36 +284,19 @@ public class ThreadSafeClientConnManager implements ClientConnectionManager {
* @return the total number of pooled connections
*/
public int getConnectionsInPool() {
connectionPool.poolLock.lock();
try {
return connectionPool.numConnections;
} finally {
connectionPool.poolLock.unlock();
}
return pool.getConnectionsInPool();
}
public void closeIdleConnections(long idleTimeout, TimeUnit tunit) {
if (log.isDebugEnabled()) {
log.debug("Closing connections idle for " + idleTimeout + " " + tunit);
}
connectionPool.poolLock.lock();
try {
connectionPool.closeIdleConnections(idleTimeout, tunit);
connectionPool.deleteClosedConnections();
} finally {
connectionPool.poolLock.unlock();
}
pool.closeIdleConnections(idleTimeout, tunit);
}
public void closeExpiredConnections() {
log.debug("Closing expired connections");
connectionPool.poolLock.lock();
try {
connectionPool.closeExpiredConnections();
connectionPool.deleteClosedConnections();
} finally {
connectionPool.poolLock.unlock();
}
pool.closeExpiredConnections();
}
/**

2
httpclient/src/test/java/org/apache/http/impl/conn/tsccm/TestSpuriousWakeup.java
View File

@ -134,7 +134,7 @@ public class TestSpuriousWakeup {
Thread.State.WAITING, gct.getState());
// get access to the objects we need
Lock lck = mgr.extendedCPBR.poolLock;
Lock lck = mgr.extendedCPBR.getLock();
Condition cnd = mgr.extendedCPBR.newestWT.getCondition();
// Now trigger spurious wakeups. We'll do it several times