diff --git a/pom.xml b/pom.xml index 7358781ca..5ad4142d6 100644 --- a/pom.xml +++ b/pom.xml @@ -129,7 +129,7 @@ mysql mysql-connector-java - 5.1.15 + 5.1.27 test diff --git a/src/main/java/com/avaje/ebean/config/DataSourceConfig.java b/src/main/java/com/avaje/ebean/config/DataSourceConfig.java index e2bde6dbe..5e55e4065 100644 --- a/src/main/java/com/avaje/ebean/config/DataSourceConfig.java +++ b/src/main/java/com/avaje/ebean/config/DataSourceConfig.java @@ -36,6 +36,8 @@ public class DataSourceConfig { private int heartbeatFreqSecs = 30; + private int heartbeatTimeoutSeconds = 3; + private boolean captureStackTrace; private int maxStackTraceSize = 5; @@ -44,6 +46,8 @@ public class DataSourceConfig { private int maxInactiveTimeSecs = 720; + private int maxAgeMinutes = 0; + private int trimPoolFreqSecs = 59; private int pstmtCacheSize = 20; @@ -55,8 +59,8 @@ public class DataSourceConfig { private String poolListener; private boolean offline; - - Map customProperties; + + protected Map customProperties; /** * Return the connection URL. @@ -196,6 +200,20 @@ public class DataSourceConfig { public void setHeartbeatFreqSecs(int heartbeatFreqSecs) { this.heartbeatFreqSecs = heartbeatFreqSecs; } + + /** + * Return the heart beat timeout in seconds. + */ + public int getHeartbeatTimeoutSeconds() { + return heartbeatTimeoutSeconds; + } + + /** + * Set the heart beat timeout in seconds. + */ + public void setHeartbeatTimeoutSeconds(int heartbeatTimeoutSeconds) { + this.heartbeatTimeoutSeconds = heartbeatTimeoutSeconds; + } /** * Return true if a stack trace should be captured when obtaining a connection @@ -311,6 +329,23 @@ public class DataSourceConfig { return maxInactiveTimeSecs; } + /** + * Return the maximum age a connection is allowed to be before it is closed. + *

+ * This can be used to close really old connections. + *

+ */ + public int getMaxAgeMinutes() { + return maxAgeMinutes; + } + + /** + * Set the maximum age a connection can be in minutes. + */ + public void setMaxAgeMinutes(int maxAgeMinutes) { + this.maxAgeMinutes = maxAgeMinutes; + } + /** * Set the time in seconds a connection can be idle after which it can be * trimmed from the pool. @@ -380,7 +415,7 @@ public class DataSourceConfig { public void setOffline(boolean offline) { this.offline = offline; } - + /** * Return a map of custom properties for the jdbc driver connection. */ @@ -420,6 +455,7 @@ public class DataSourceConfig { this.leakTimeMinutes = properties.getInt(prefix + "leakTimeMinutes", 30); this.maxInactiveTimeSecs = properties.getInt(prefix + "maxInactiveTimeSecs", 720); this.trimPoolFreqSecs = properties.getInt(prefix + "trimPoolFreqSecs", 59); + this.maxAgeMinutes = properties.getInt(prefix + "maxAgeMinutes", 0); this.minConnections = properties.getInt(prefix + "minConnections", 0); this.maxConnections = properties.getInt(prefix + "maxConnections", 20); @@ -429,6 +465,7 @@ public class DataSourceConfig { this.waitTimeoutMillis = properties.getInt(prefix + "waitTimeout", 1000); this.heartbeatSql = properties.get(prefix + "heartbeatSql", null); + this.heartbeatTimeoutSeconds = properties.getInt(prefix + "heartbeatTimeoutSeconds", 3); this.poolListener = properties.get(prefix + "poolListener", null); this.offline = properties.getBoolean(prefix + "offline", false); diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java index 2bf4dc6fe..6aa762474 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java @@ -410,32 +410,10 @@ public class DefaultServerFactory implements BootupEbeanManager { return null; } - if (dsConfig.getHeartbeatSql() == null) { - // use default heartbeatSql from the DatabasePlatform - String heartbeatSql = getHeartbeatSql(dsConfig.getDriver()); - dsConfig.setHeartbeatSql(heartbeatSql); - } - DataSourceAlert notify = new SimpleDataSourceAlert(); return new DataSourcePool(notify, config.getName(), dsConfig); } - /** - * Return a heartbeatSql depending on the jdbc driver name. - */ - private String getHeartbeatSql(String driver) { - if (driver != null) { - String d = driver.toLowerCase(); - if (d.contains("oracle")) { - return "select 'x' from dual"; - } - if (d.contains(".h2.") || d.contains(".mysql.") || d.contains("postgre")) { - return "select 1"; - } - } - return null; - } - /** * Check the autoCommit and Transaction Isolation levels of the DataSource. *

diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/BusyConnectionBuffer.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/BusyConnectionBuffer.java index 98b55d682..98b50346b 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/BusyConnectionBuffer.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/BusyConnectionBuffer.java @@ -1,9 +1,6 @@ package com.avaje.ebeaninternal.server.lib.sql; -import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; -import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,7 +50,7 @@ class BusyConnectionBuffer { /** * We can only grow (not shrink) the capacity. */ - private void setCapacity(int newCapacity) { + protected void setCapacity(int newCapacity) { if (newCapacity > slots.length){ PooledConnection[] current = this.slots; this.slots = new PooledConnection[newCapacity]; @@ -74,12 +71,7 @@ class BusyConnectionBuffer { } protected boolean isEmpty() { - for (int i = 0; i < slots.length; i++) { - if (slots[i] != null) { - return false; - } - } - return true; + return size == 0; } protected int add(PooledConnection pc){ @@ -120,22 +112,70 @@ class BusyConnectionBuffer { } /** - * Get a shallow read only List of the busy connections. - *

- * Note that the {@link #remove(PooledConnection)} MUST be used to remove PooledConnections. - *

- * @return + * Close connections that should be considered leaked. */ - protected List getShallowCopy() { - ArrayList tmp = new ArrayList(); - for (int i = 0; i < slots.length; i++) { - if (slots[i] != null){ - tmp.add(slots[i]); + protected void closeBusyConnections(long leakTimeMinutes) { + + long olderThanTime = System.currentTimeMillis() - (leakTimeMinutes*60000); + + logger.debug("Closing busy connections using leakTimeMinutes {}", leakTimeMinutes); + + for (int i = 0; i < slots.length; i++) { + if (slots[i] != null){ + //tmp.add(slots[i]); + PooledConnection pc = slots[i]; + if (pc.isLongRunning() || pc.getLastUsedTime() > olderThanTime) { + // PooledConnection has been used recently or + // expected to be longRunning so not closing... + } else { + slots[i] = null; + --size; + closeBusyConnection(pc); } } - return Collections.unmodifiableList(tmp); + } } + private void closeBusyConnection(PooledConnection pc) { + try { + + logger.warn("DataSourcePool closing busy connection? "+pc.getFullDescription()); + System.out.println("CLOSING busy connection: "+pc.getFullDescription()); + + pc.closeConnectionFully(false); + + } catch (Exception ex) { + // this should never actually happen + logger.error("Error when closing potentially leaked connection "+pc.getDescription(), ex); + } + } + + /** + * Returns information describing connections that are currently being used. + */ + protected String getBusyConnectionInformation(boolean toLogger) { + + if (toLogger) { + logger.info("Dumping [{}] busy connections: (Use datasource.xxx.capturestacktrace=true ... to get stackTraces)", size()); + } + + StringBuilder sb = new StringBuilder(); + + for (int i = 0; i < slots.length; i++) { + if (slots[i] != null){ + PooledConnection pc = slots[i]; + if (toLogger) { + logger.info("Busy Connection - {}", pc.getFullDescription()); + } else { + sb.append(pc.getFullDescription()).append("\r\n"); + } + } + } + + return sb.toString(); + } + + /** * Return the position of the next empty slot. */ diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java index 0dff23448..de80b9c7b 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java @@ -76,8 +76,11 @@ public class DataSourcePool implements DataSource { private final String heartbeatsql; private final int heartbeatFreqSecs; + + private final int heartbeatTimeoutSeconds; + - private final int trimPoolFreqSecs; + private final long trimPoolFreqMillis; /** * The transaction isolation level as per java.sql.Connection. @@ -89,6 +92,17 @@ public class DataSourcePool implements DataSource { */ private final boolean autoCommit; + /** + * Max idle time in millis. + */ + private final int maxInactiveMillis; + + /** + * Max age a connection is allowed in millis. + * A value of 0 means no limit (no trimming based on max age). + */ + private final long maxAgeMillis; + /** * Flag set to true to capture stackTraces (can be expensive). */ @@ -138,17 +152,12 @@ public class DataSourcePool implements DataSource { /** * The time a thread will wait for a connection to become available. */ - private int waitTimeoutMillis; + private final int waitTimeoutMillis; /** * The size of the preparedStatement cache; */ private int pstmtCacheSize; - - /** - * By default trim connections that are inactive for longer than this time. - */ - private int maxInactiveTimeSecs; private final PooledConnectionQueue queue; @@ -169,8 +178,9 @@ public class DataSourcePool implements DataSource { this.autoCommit = false; this.transactionIsolation = params.getIsolationLevel(); - - this.maxInactiveTimeSecs = params.getMaxInactiveTimeSecs(); + + this.maxInactiveMillis = 1000 * params.getMaxInactiveTimeSecs(); + this.maxAgeMillis = 60000 * params.getMaxAgeMinutes(); this.leakTimeMinutes = params.getLeakTimeMinutes(); this.captureStackTrace = params.isCaptureStackTrace(); this.maxStackTraceSize = params.getMaxStackTraceSize(); @@ -183,7 +193,8 @@ public class DataSourcePool implements DataSource { this.waitTimeoutMillis = params.getWaitTimeoutMillis(); this.heartbeatsql = params.getHeartbeatSql(); this.heartbeatFreqSecs = params.getHeartbeatFreqSecs(); - this.trimPoolFreqSecs = params.getTrimPoolFreqSecs(); + this.heartbeatTimeoutSeconds = params.getHeartbeatTimeoutSeconds(); + this.trimPoolFreqMillis = 1000 * params.getTrimPoolFreqSecs(); queue = new PooledConnectionQueue(this); @@ -321,12 +332,11 @@ public class DataSourcePool implements DataSource { private void notifyDataSourceIsDown(SQLException ex) { if (!dataSourceDownAlertSent) { - logger.error("FATAL: DataSourcePool [" + name + "] is down!!!", ex); + logger.error("FATAL: DataSourcePool [" + name + "] is down or has network error!!!", ex); if (notify != null) { notify.dataSourceDown(name); } dataSourceDownAlertSent = true; - } if (dataSourceUp) { reset(); @@ -370,6 +380,20 @@ public class DataSourcePool implements DataSource { return heartbeatRunnable; } + /** + * Trim connections (in the free list) based on idle time and maximum age. + */ + private void trimIdleConnections() { + if (System.currentTimeMillis() > (lastTrimTime + trimPoolFreqMillis)) { + try { + queue.trim(maxInactiveMillis, maxAgeMillis); + lastTrimTime = System.currentTimeMillis(); + } catch (Exception e) { + logger.error("Error trying to trim idle connections", e); + } + } + } + /** * Check the dataSource is up. Trim connections. *

@@ -378,17 +402,19 @@ public class DataSourcePool implements DataSource { *

*/ public void checkDataSource() { + + // first trim idle connections + trimIdleConnections(); + Connection conn = null; try { // Get a connection from the pool and test it conn = getConnection(); - testConnection(conn); - - notifyDataSourceIsUp(); - - if (System.currentTimeMillis() > (lastTrimTime + (trimPoolFreqSecs * 1000))) { - queue.trim(maxInactiveTimeSecs); - lastTrimTime = System.currentTimeMillis(); + if (testConnection(conn)) { + notifyDataSourceIsUp(); + + } else { + notifyDataSourceIsDown(null); } } catch (SQLException ex) { @@ -491,33 +517,38 @@ public class DataSourcePool implements DataSource { return waitTimeoutMillis; } - /** - * Set the time after which inactive connections are trimmed. - */ - public void setMaxInactiveTimeSecs(int maxInactiveTimeSecs) { - this.maxInactiveTimeSecs = maxInactiveTimeSecs; - } - /** * Return the time after which inactive connections are trimmed. */ - public int getMaxInactiveTimeSecs() { - return maxInactiveTimeSecs; + public int getMaxInactiveMillis() { + return maxInactiveMillis; + } + + /** + * Return the maximum age a connection is allowed to be before it is trimmed + * out of the pool. This value can be 0 which means there is no maximum age. + */ + public long getMaxAgeMillis() { + return maxAgeMillis; } - private void testConnection(Connection conn) throws SQLException { + private boolean testConnection(Connection conn) throws SQLException { if (heartbeatsql == null) { - return; + return conn.isValid(heartbeatTimeoutSeconds); } Statement stmt = null; ResultSet rset = null; try { - // It should only error IF the DataSource is down ? (or a network - // issue?) + // It should only error IF the DataSource is down or a network issue stmt = conn.createStatement(); + if (heartbeatTimeoutSeconds > 0) { + stmt.setQueryTimeout(heartbeatTimeoutSeconds); + } rset = stmt.executeQuery(heartbeatsql); conn.commit(); + + return true; } finally { try { @@ -543,13 +574,7 @@ public class DataSourcePool implements DataSource { */ protected boolean validateConnection(PooledConnection conn) { try { - if (heartbeatsql == null) { - logger.debug("Can not test connection as heartbeatsql is not set"); - return false; - } - - testConnection(conn); - return true; + return testConnection(conn); } catch (Exception e) { logger.warn("heartbeatsql test failed on connection[" + conn.getName() + "]"); @@ -647,7 +672,6 @@ public class DataSourcePool implements DataSource { notifyDataSourceIsDown(ex); throw ex; } - } /** diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/FreeConnectionBuffer.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/FreeConnectionBuffer.java index 6186aebd8..8fede2fc2 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/FreeConnectionBuffer.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/FreeConnectionBuffer.java @@ -1,179 +1,106 @@ package com.avaje.ebeaninternal.server.lib.sql; import java.util.ArrayList; +import java.util.Iterator; +import java.util.LinkedList; import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.avaje.ebeaninternal.server.lib.sql.PooledConnectionStatistics.LoadValues; /** * A buffer designed especially to hold free pooled connections. *

- * It is circular in nature. - *

- *

* All thread safety controlled externally (by PooledConnectionQueue). *

- * - * @author rbygrave - * */ class FreeConnectionBuffer { - /** - * The buffer itself. - */ - private PooledConnection[] conns; + private static final Logger logger = LoggerFactory.getLogger(FreeConnectionBuffer.class); + + /** + * Buffer oriented for add and remove. + */ + private final LinkedList freeBuffer = new LinkedList(); - /** - * The position in the buffer where the next connection is removed from. - */ - private int removeIndex; + protected FreeConnectionBuffer() { + } - /** - * Position in the buffer where the next connection is added to. - */ - private int addIndex; + protected int size() { + return freeBuffer.size(); + } - /** - * The current number of connections in the buffer - */ - private int size; + protected boolean isEmpty() { + return freeBuffer.isEmpty(); + } + + /** + * Add connection to the free list. + */ + protected void add(PooledConnection pc) { + freeBuffer.addLast(pc); + } + + /** + * Remove a connection from the free list. + */ + protected PooledConnection remove() { + return freeBuffer.removeFirst(); + } + + /** + * Close all connections in this buffer. + */ + protected void closeAll(boolean logErrors) { + + // create a temporary list + List tempList = new ArrayList(freeBuffer.size()); - protected FreeConnectionBuffer(int capacity) { - this.conns = new PooledConnection[capacity]; + // add all the connections into it + for (PooledConnection c : freeBuffer) { + tempList.add(c); } - protected int getCapacity() { - return conns.length; - } + // clear the buffer (in case it takes some time to close these connections). + freeBuffer.clear(); - protected int size() { - return size; - } - - protected boolean isEmpty() { - return size == 0; - } - - /** - * Add at connection. - */ - protected void add(PooledConnection pc) { - if (conns[addIndex] != null) { - throw new IllegalStateException("Buffer slot ["+addIndex+"] already full?"); - } - conns[addIndex] = pc; - addIndex = inc(addIndex); - ++size; + logger.debug("... closing all {} connections from the free list with logErrors: {}", tempList.size(), logErrors); + for (int i = 0; i < tempList.size(); i++) { + PooledConnection pooledConnection = tempList.get(i); + logger.debug("... closing {} of {} connections from the free list", i, tempList.size()); + pooledConnection.closeConnectionFully(logErrors); } + } + + /** + * Trim any inactive connections that have not been used since usedSince. + */ + protected int trim(long usedSince, long createdSince) { - /** - * Close all connections in this buffer. - */ - protected void closeAll(boolean logErrors) { - - final PooledConnection[] items = this.conns; - - this.conns = new PooledConnection[items.length]; - this.size = 0; - this.removeIndex = 0; - this.addIndex = 0; + int trimCount = 0; - for (int i = 0; i < items.length; i++) { - PooledConnection c = items[i]; - if (c != null) { - c.closeConnectionFully(logErrors); - } + Iterator iterator = freeBuffer.iterator(); + while (iterator.hasNext()) { + PooledConnection pooledConnection = iterator.next(); + if (pooledConnection.shouldTrim(usedSince, createdSince)) { + iterator.remove(); + pooledConnection.closeConnectionFully(true); + trimCount++; } } - - /** - * Remove a connection at current remove position. - */ - protected PooledConnection remove() { - final PooledConnection[] items = this.conns; - PooledConnection pc = items[removeIndex]; - items[removeIndex] = null; - removeIndex = inc(removeIndex); - --size; - return pc; - } - - /** - * Trim any inactive connections that have not been used since usedSince. - */ - protected int trim(long usedSince) { - - int trimCount = 0; - for (int i = 0; i < conns.length; i++) { - if (conns[i] != null){ - if (conns[i].getLastUsedTime() < usedSince) { - trimCount++; - conns[i].closeConnectionFully(true); - conns[i] = null; - --size; - } - } - } - - return trimCount; - } - - /** - * Collect the load statistics from all the free connections. - */ - protected void collectStatistics(LoadValues values, boolean reset) { - - for (int i = 0; i < conns.length; i++) { - if (conns[i] != null){ - values.plus(conns[i].getStatistics().getValues(reset)); - } - } - } - - /** - * Return a shallow copy of the free connections. - */ - private List getShallowCopy() { - - List copy = new ArrayList(conns.length); - for (int i = 0; i < conns.length; i++) { - if (conns[i] != null){ - copy.add(conns[i]); - } - } - return copy; - } - /** - * Increase the capacity of the buffer. This is a relatively expensive - * operation but should occur very infrequently. - */ - protected void setCapacity(int newCapacity) { - - if (newCapacity > conns.length){ - - List copy = getShallowCopy(); - - // reset to empty state - this.removeIndex = 0; - this.addIndex = 0; - this.size = 0; + return trimCount; + } - this.conns = new PooledConnection[newCapacity]; + /** + * Collect the load statistics from all the free connections. + */ + protected void collectStatistics(LoadValues values, boolean reset) { - // add the connections back from the copy - for (int i = 0; i < copy.size(); i++) { - add(copy.get(i)); - } - } + for (PooledConnection c : freeBuffer) { + values.plus(c.getStatistics().getValues(reset)); } - - /** - * Circularly increment i. - */ - private final int inc(int i) { - return (++i == conns.length)? 0 : i; - } - + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java index 9712d5482..cd6ade3e0 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java @@ -10,14 +10,13 @@ import java.sql.Savepoint; import java.sql.Statement; import java.util.ArrayList; import java.util.Arrays; -import java.util.Iterator; import java.util.Map; -import com.avaje.ebeaninternal.jdbc.ConnectionDelegator; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.avaje.ebeaninternal.jdbc.ConnectionDelegator; + /** * Is a connection that belongs to a DataSourcePool. * @@ -40,8 +39,23 @@ public class PooledConnection extends ConnectionDelegator { private static final Logger logger = LoggerFactory.getLogger(PooledConnection.class); - private static String IDLE_CONNECTION_ACCESSED_ERROR = "Pooled Connection has been accessed whilst idle in the pool, via method: "; + private static final String IDLE_CONNECTION_ACCESSED_ERROR = "Pooled Connection has been accessed whilst idle in the pool, via method: "; + /** + * Marker for when connection is closed due to exceeding the max allowed age. + */ + private static final String REASON_MAXAGE = "maxAge"; + + /** + * Marker for when connection is closed due to exceeding the max inactive time. + */ + private static final String REASON_IDLE = "idleTime"; + + /** + * Marker for when the connection is closed due to a reset. + */ + private static final String REASON_RESET = "reset"; + /** * Set when connection is idle in the pool. In general when in the pool the * connection should not be modified. @@ -85,11 +99,21 @@ public class PooledConnection extends ConnectionDelegator { private final Object pstmtMonitor = new Object(); + /** + * Helper for statistics collection. + */ + private final PooledConnectionStatistics stats = new PooledConnectionStatistics(); + /** * The status of the connection. IDLE, ACTIVE or ENDED. */ private int status = STATUS_IDLE; + /** + * The reason for a connection closing. + */ + private String closeReason; + /** * Set this to true if the connection will be busy for a long time. *

@@ -117,8 +141,6 @@ public class PooledConnection extends ConnectionDelegator { private long exeStartNanos; - private final PooledConnectionStatistics stats = new PooledConnectionStatistics(); - /** * The last statement executed by this connection. */ @@ -273,7 +295,7 @@ public class PooledConnection extends ConnectionDelegator { } if (logger.isDebugEnabled()) { - logger.debug("Closing Connection[{}] slot[{}] Stats: {} , PstmtStats: {} ", name, slotId, stats.getValues(false), pstmtCache.getDescription()); + logger.debug("Closing Connection[{}] slot[{}] reason[{}] stats: {} , pstmtStats: {} ", name, slotId, closeReason, stats.getValues(false), pstmtCache.getDescription()); } try { @@ -290,11 +312,9 @@ public class PooledConnection extends ConnectionDelegator { } try { - Iterator psi = pstmtCache.values().iterator(); - while (psi.hasNext()) { - ExtendedPreparedStatement ps = (ExtendedPreparedStatement) psi.next(); - ps.closeDestroy(); - } + for (ExtendedPreparedStatement ps : pstmtCache.values()) { + ps.closeDestroy(); + } } catch (SQLException ex) { if (logErrors) { @@ -335,8 +355,7 @@ public class PooledConnection extends ConnectionDelegator { } } - public Statement createStatement(int resultSetType, int resultSetConcurreny) - throws SQLException { + public Statement createStatement(int resultSetType, int resultSetConcurreny) throws SQLException { if (status == STATUS_IDLE) { throw new SQLException(IDLE_CONNECTION_ACCESSED_ERROR + "createStatement()"); } @@ -388,8 +407,7 @@ public class PooledConnection extends ConnectionDelegator { private PreparedStatement prepareStatement(String sql, boolean useFlag, int flag, String cacheKey) throws SQLException { if (status == STATUS_IDLE) { - String m = IDLE_CONNECTION_ACCESSED_ERROR + "prepareStatement()"; - throw new SQLException(m); + throw new SQLException(IDLE_CONNECTION_ACCESSED_ERROR + "prepareStatement()"); } try { synchronized (pstmtMonitor) { @@ -418,8 +436,8 @@ public class PooledConnection extends ConnectionDelegator { } } - public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurreny) - throws SQLException { + public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurreny) throws SQLException { + if (status == STATUS_IDLE) { throw new SQLException(IDLE_CONNECTION_ACCESSED_ERROR + "prepareStatement()"); } @@ -518,6 +536,7 @@ public class PooledConnection extends ConnectionDelegator { } catch (Exception ex) { // the connection is BAD, close it and test the pool + logger.warn("Error when trying to return connection to pool, closing fully.", ex); closeConnectionFully(false); pool.checkDataSource(); } @@ -538,7 +557,7 @@ public class PooledConnection extends ConnectionDelegator { try { if (connection != null && !connection.isClosed()) { // connect leak? - logger.warn("Closing Connection[" + getName() + "] on finalize()."); + logger.warn("Closing Connection on finalize() - {}", getFullDescription()); closeConnectionFully(false); } } catch (Exception e) { @@ -547,6 +566,45 @@ public class PooledConnection extends ConnectionDelegator { super.finalize(); } + /** + * Return true if the connection is too old. + */ + public boolean exceedsMaxAge(long maxAgeMillis) { + if (maxAgeMillis > 0 && (creationTime < (System.currentTimeMillis() - maxAgeMillis))){ + this.closeReason = REASON_MAXAGE; + return true; + } + return false; + } + + public boolean shouldTrimOnReturn(long lastResetTime, long maxAgeMillis) { + if (creationTime <= lastResetTime) { + this.closeReason = REASON_RESET; + return true; + } + if (exceedsMaxAge(maxAgeMillis)) { + return true; + } + return false; + } + + /** + * Return true if the connection has been idle for too long or is too old. + */ + public boolean shouldTrim(long usedSince, long createdSince) { + if (lastUseTime < usedSince) { + // been idle for too long so trim it + this.closeReason = REASON_IDLE; + return true; + } + if (createdSince > 0 && createdSince > creationTime) { + // exceeds max age so trim it + this.closeReason = REASON_MAXAGE; + return true; + } + return false; + } + /** * Return the time the connection was passed to the client code. *

diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java index 1076f9ad0..e355394d1 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java @@ -1,7 +1,6 @@ package com.avaje.ebeaninternal.server.lib.sql; import java.sql.SQLException; -import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; @@ -55,10 +54,12 @@ public class PooledConnectionQueue { private int connectionId; - private long waitTimeoutMillis; - - private long leakTimeMinutes; + private final long waitTimeoutMillis; + private final long leakTimeMinutes; + + private final long maxAgeMillis; + private int warningSize; private int maxSize; @@ -103,9 +104,10 @@ public class PooledConnectionQueue { this.warningSize = pool.getWarningSize(); this.waitTimeoutMillis = pool.getWaitTimeoutMillis(); this.leakTimeMinutes = pool.getLeakTimeMinutes(); - + this.maxAgeMillis = pool.getMaxAgeMillis(); + this.busyList = new BusyConnectionBuffer(maxSize, 20); - this.freeList = new FreeConnectionBuffer(maxSize); + this.freeList = new FreeConnectionBuffer(); this.lock = new ReentrantLock(false); this.notEmpty = lock.newCondition(); @@ -191,7 +193,7 @@ public class PooledConnectionQueue { if (maxSize < this.minSize){ throw new IllegalArgumentException("maxSize "+maxSize+" < minSize "+this.minSize); } - freeList.setCapacity(maxSize); + this.busyList.setCapacity(maxSize); this.maxSize = maxSize; } finally { lock.unlock(); @@ -244,8 +246,9 @@ public class PooledConnectionQueue { if (!busyList.remove(c)) { logger.error("Connection [{}] not found in BusyList? ", c); } - if (c.getCreationTime() <= lastResetTime) { + if (c.shouldTrimOnReturn(lastResetTime, maxAgeMillis)) { c.closeConnectionFully(false); + } else { freeList.add(c); notEmpty.signal(); @@ -310,9 +313,9 @@ public class PooledConnectionQueue { PooledConnection c = pool.createConnectionForQueue(connectionId++); int busySize = registerBusyConnection(c); - String msg = "DataSourcePool [" + name + "] grow; id["+c.getName()+"] busy["+busySize+"] max["+maxSize+"]"; - logger.debug(msg); - + if (logger.isDebugEnabled()) { + logger.debug("DataSourcePool [{}] grow; id[{}] busy[{}] max[{}]", name, c.getName(), busySize, maxSize); + } checkForWarningSize(); return c; } @@ -372,7 +375,7 @@ public class PooledConnectionQueue { doingShutdown = true; Status status = createStatus(); DataSourcePoolStatistics statistics = pool.getStatistics(false); - logger.debug("DataSourcePool [" + name + "] shutdown {} - Statistics {}", status, statistics); + logger.debug("DataSourcePool [{}] shutdown {} - Statistics {}", name, status, statistics); closeFreeConnections(true); @@ -398,7 +401,7 @@ public class PooledConnectionQueue { lock.lock(); try { Status status = createStatus(); - logger.info("Reseting DataSourcePool [" + name + "] "+status); + logger.info("Reseting DataSourcePool [{}] {}", name, status); lastResetTime = System.currentTimeMillis(); closeFreeConnections(false); @@ -411,12 +414,16 @@ public class PooledConnectionQueue { } } - public void trim(int maxInactiveTimeSecs) throws SQLException { + public void trim(long maxInactiveMillis, long maxAgeMillis) { final ReentrantLock lock = this.lock; lock.lock(); try { - if (trimInactiveConnections(maxInactiveTimeSecs) > 0) { - ensureMinimumConnections(); + if (trimInactiveConnections(maxInactiveMillis, maxAgeMillis) > 0) { + try { + ensureMinimumConnections(); + } catch (SQLException e) { + logger.error("Error trying to ensure minimum connections", e); + } } } finally { lock.unlock(); @@ -426,11 +433,12 @@ public class PooledConnectionQueue { /** * Trim connections that have been not used for some time. */ - private int trimInactiveConnections(int maxInactiveTimeSecs) { + private int trimInactiveConnections(long maxInactiveMillis, long maxAgeMillis) { - long usedSince = System.currentTimeMillis() - (maxInactiveTimeSecs * 1000); - - int trimedCount = freeList.trim(usedSince); + long usedSince = System.currentTimeMillis() - maxInactiveMillis; + long createdSince = (maxAgeMillis == 0) ? 0 : System.currentTimeMillis() - maxAgeMillis; + + int trimedCount = freeList.trim(usedSince, createdSince); if (trimedCount > 0) { logger.debug("DataSourcePool [{}] trimmed [{}] inactive connections. New size[{}]", name, trimedCount, totalConnections()); } @@ -467,42 +475,11 @@ public class PooledConnectionQueue { final ReentrantLock lock = this.lock; lock.lock(); try { - - long olderThanTime = System.currentTimeMillis() - (leakTimeMinutes*60000); - - List copy = busyList.getShallowCopy(); - - logger.debug("Closing busy connections using leakTimeMinutes {}", leakTimeMinutes); - - for (int i = 0; i < copy.size(); i++) { - PooledConnection pc = copy.get(i); - if (pc.isLongRunning() || pc.getLastUsedTime() > olderThanTime) { - // PooledConnection has been used recently or - // expected to be longRunning so not closing... - } else { - busyList.remove(pc); - closeBusyConnection(pc, leakTimeMinutes); - } - } - + busyList.closeBusyConnections(leakTimeMinutes); } finally { lock.unlock(); } } - - private void closeBusyConnection(PooledConnection pc, long leakMinutes) { - try { - - logger.warn("DataSourcePool closing busy connection? "+pc.getFullDescription()); - System.out.println("CLOSING busy connection: "+pc.getFullDescription()); - - pc.closeConnectionFully(false); - - } catch (Exception ex) { - // this should never actually happen - logger.error("Error when closing potentially leaked connection "+pc.getDescription(), ex); - } - } /** * As the pool grows it gets closer to the maxConnections limit. We can send @@ -544,25 +521,8 @@ public class PooledConnectionQueue { lock.lock(); try { - StringBuilder sb = new StringBuilder(); - - List copy = busyList.getShallowCopy(); - - if (toLogger) { - logger.info("Dumping [{}] busy connections: (Use datasource.xxx.capturestacktrace=true ... to get stackTraces)", copy.size()); - } - - for (int i = 0; i < copy.size(); i++) { - PooledConnection pc = copy.get(i); - if (toLogger) { - logger.info("Busy Connection - {}", pc.getFullDescription()); - } else { - sb.append(pc.getFullDescription()).append("\r\n"); - } - } - - return sb.toString(); - + return busyList.getBusyConnectionInformation(toLogger); + } finally { lock.unlock(); } diff --git a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMax.java b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMax.java index 26fe97c91..c392515a9 100644 --- a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMax.java +++ b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMax.java @@ -7,7 +7,6 @@ import java.sql.ResultSet; import org.junit.Test; import com.avaje.ebean.BaseTestCase; -import com.avaje.ebean.Ebean; import com.avaje.ebean.config.DataSourceConfig; import com.avaje.ebeaninternal.server.core.DefaultBackgroundExecutor; import com.avaje.ebeaninternal.server.lib.sql.DataSourcePool.Status; @@ -22,43 +21,45 @@ public class TestDataSourceMax extends BaseTestCase { if (skipThisTest) { return; } - - Ebean.getServer(null); - String name = "h2"; + String name = "mysql"; DataSourceConfig dsConfig = new DataSourceConfig(); dsConfig.loadSettings(name); dsConfig.setMinConnections(2); - dsConfig.setMaxConnections(8); + dsConfig.setMaxConnections(25); dsConfig.setWaitTimeoutMillis(30000); dsConfig.setCaptureStackTrace(true); DataSourcePool pool = new DataSourcePool(null, name, dsConfig); - // Assert.assertEquals(3, pool.getMaxSize()); - - DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(10, 2, 180, 30, "testDs"); + + //pool.checkDataSource(); + +// if (true) { +// pool.shutdown(false); +// return; +// } + + DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(1, 2, 180, 30, "testDs"); try { for (int i = 0; i < 12; i++) { // Thread.sleep(10*i); - bg.execute(new ConnRunner(pool, 4000)); + bg.execute(new ConnRunner(pool, 4000, i)); } System.out.println("main thread sleep ... " + pool.getStatus(false)); Thread.sleep(10000); + pool.getStatistics(true); + + Thread.sleep(30000); + Status status = pool.getStatus(false); System.out.println(status); pool.shutdown(false); - // this dumpOrder was for 3 vectors used in PooledConnectionQueue - // that logged the order of wait, notify and obtain events - // I have remove that code. - - // String s = pool.dumpOrder(); - // System.err.println(s); } catch (Exception e) { e.printStackTrace(); @@ -70,24 +71,38 @@ public class TestDataSourceMax extends BaseTestCase { final DataSourcePool pool; final long sleepMillis; + final int position; - ConnRunner(DataSourcePool pool, long sleepMillis) { + ConnRunner(DataSourcePool pool, long sleepMillis, int position) { this.pool = pool; this.sleepMillis = sleepMillis; + this.position = position; } + private void waitSomeTime(long count) { + try { + System.out.println(position+" sleep " + sleepMillis+" count:"+count); + Thread.sleep(sleepMillis); + System.out.println(position+" sleep done"); + } catch (InterruptedException e){ + throw new RuntimeException(e); + } + } + public void run() { Connection connection = null; PreparedStatement pstmt = null; ResultSet rset = null; + long count = -1; try { connection = pool.getConnection(); pstmt = connection.prepareStatement("select count(*) from o_customer"); rset = pstmt.executeQuery(); - - System.out.println("sleep " + sleepMillis); - Thread.sleep(sleepMillis); - System.out.println("sleep done"); + + while (rset.next()) { + // do nothing actually + count = rset.getLong(1); + } } catch (Exception ex) { ex.printStackTrace(); @@ -113,6 +128,7 @@ public class TestDataSourceMax extends BaseTestCase { } } } + waitSomeTime(count); } } } diff --git a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMaxWithEntity.java b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMaxWithEntity.java new file mode 100644 index 000000000..b4d136a7c --- /dev/null +++ b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestDataSourceMaxWithEntity.java @@ -0,0 +1,70 @@ +package com.avaje.ebeaninternal.server.lib.sql; + +import org.junit.Test; + +import com.avaje.ebean.BaseTestCase; +import com.avaje.ebean.Ebean; +import com.avaje.ebean.EbeanServer; +import com.avaje.ebeaninternal.server.core.DefaultBackgroundExecutor; +import com.avaje.tests.model.basic.Customer; + +public class TestDataSourceMaxWithEntity extends BaseTestCase { + + @Test + public void test() { + + boolean skipThisTest = true; + + if (skipThisTest) { + return; + } + + EbeanServer server = Ebean.getServer(null); + + + DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(1, 2, 180, 30, "testDs"); + + try { + for (int i = 0; i < 12; i++) { + // Thread.sleep(10*i); + bg.execute(new ConnRunner(server, 4000, i)); + } + + System.out.println("main thread sleep ... "); + + + Thread.sleep(30000); + + server.shutdown(true, false); + + } catch (Exception e) { + e.printStackTrace(); + } + + } + + private static class ConnRunner implements Runnable { + + final EbeanServer server; + final long sleepMillis; + final int position; + + ConnRunner(EbeanServer server, long sleepMillis, int position) { + this.server = server; + this.sleepMillis = sleepMillis; + this.position = position; + } + + public void run() { + + server.find(Customer.class).findRowCount(); + try { + System.out.println(position+" sleep " + sleepMillis); + Thread.sleep(sleepMillis); + System.out.println(position+" sleep done"); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } +} diff --git a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBuffer.java b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBuffer.java index de3a451b1..85e0f1a4e 100644 --- a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBuffer.java +++ b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBuffer.java @@ -10,14 +10,13 @@ public class TestFreeBuffer extends BaseTestCase { @Test public void test() { - FreeConnectionBuffer b = new FreeConnectionBuffer(3); + FreeConnectionBuffer b = new FreeConnectionBuffer(); PooledConnection p0 = new PooledConnection("0"); PooledConnection p1 = new PooledConnection("1"); PooledConnection p2 = new PooledConnection("2"); // PooledConnection p3 = new PooledConnection("3"); - Assert.assertEquals(3, b.getCapacity()); Assert.assertEquals(0, b.size()); Assert.assertEquals(true, b.isEmpty()); diff --git a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBufferTrim.java b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBufferTrim.java index 9a3c8b293..a7a4052c2 100644 --- a/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBufferTrim.java +++ b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBufferTrim.java @@ -7,44 +7,21 @@ import org.mockito.Mockito; import com.avaje.ebean.BaseTestCase; public class TestFreeBufferTrim extends BaseTestCase { - - @Test - public void test() { - - FreeConnectionBuffer b = new FreeConnectionBuffer(3); - Assert.assertEquals(0, b.size()); - - PooledConnection p0 = Mockito.mock(PooledConnection.class); - PooledConnection p1 = Mockito.mock(PooledConnection.class); - PooledConnection p2 = Mockito.mock(PooledConnection.class); - b.add(p0); - b.add(p1); - b.add(p2); - - Assert.assertEquals(3, b.size()); - - // add 1 second as this is going to fast for System.currentTimeMillis() - long now = System.currentTimeMillis()+1000; - int trimCount = b.trim(now); - - Assert.assertEquals(0, b.size()); - Assert.assertEquals(3, trimCount); - } @Test public void testWithTime() { - FreeConnectionBuffer b = new FreeConnectionBuffer(3); + FreeConnectionBuffer b = new FreeConnectionBuffer(); Assert.assertEquals(0, b.size()); PooledConnection p0 = Mockito.mock(PooledConnection.class); - Mockito.when(p0.getLastUsedTime()).thenReturn(1000l); + Mockito.when(p0.shouldTrim(1500, 0)).thenReturn(true); PooledConnection p1 = Mockito.mock(PooledConnection.class); - Mockito.when(p1.getLastUsedTime()).thenReturn(2000l); - + Mockito.when(p1.shouldTrim(1500, 0)).thenReturn(true); + PooledConnection p2 = Mockito.mock(PooledConnection.class); - Mockito.when(p2.getLastUsedTime()).thenReturn(1100l); + Mockito.when(p2.shouldTrim(1500, 0)).thenReturn(false); b.add(p0); b.add(p1); @@ -52,7 +29,7 @@ public class TestFreeBufferTrim extends BaseTestCase { Assert.assertEquals(3, b.size()); - int trimCount = b.trim(1500); + int trimCount = b.trim(1500, 0); Assert.assertEquals(1, b.size()); Assert.assertEquals(2, trimCount); diff --git a/src/test/resources/ebean.properties b/src/test/resources/ebean.properties index b86889338..676df1c62 100644 --- a/src/test/resources/ebean.properties +++ b/src/test/resources/ebean.properties @@ -25,7 +25,7 @@ ebean.autofetch.traceUsageCollection=false ebean.ddl.generate=true -ebean.ddl.run=true +#ebean.ddl.run=true ebean.debug.sql=true @@ -80,7 +80,7 @@ ebean.cacheWarmingDelay=-1 ## DataSources (If using default Ebean DataSourceFactory) ## ------------------------------------------------------------- -datasource.default=h2 +datasource.default=mysql datasource.h2.username=sa datasource.h2.password= @@ -115,7 +115,7 @@ datasource.mysql.username=test datasource.mysql.password=test datasource.mysql.databaseUrl=jdbc:mysql://127.0.0.1:3306/test datasource.mysql.databaseDriver=com.mysql.jdbc.Driver -datasource.mysql.minConnections=1 +datasource.mysql.minConnections=2 datasource.mysql.maxConnections=25 #datasource.mysql.heartbeatsql=select count(*) from dual datasource.mysql.isolationlevel=read_committed