github/ebean
4
0
Fork 0
mirror of https://github.com/ebean-orm/ebean.git synced 2024-04-21 10:51:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support heart beat timeout, tidy up free buffer

Browse Source
This commit is contained in:
Rob Bygrave
2014-01-16 00:28:00 +13:00
parent 4900a460d1
commit 5f4789f9ba
13 changed files with 464 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/avaje/ebean/config/DataSourceConfig.java
+40 -3
View File
@@ -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<String, String> customProperties;
protected Map<String, String> 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.
* <p>
* This can be used to close really old connections.
* </p>
*/
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);
src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java
-22
View File
@@ -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.
* <p>
src/main/java/com/avaje/ebeaninternal/server/lib/sql/BusyConnectionBuffer.java
+61 -21
View File
@@ -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.
* <p>
* Note that the {@link #remove(PooledConnection)} MUST be used to remove PooledConnections.
* </p>
* @return
* Close connections that should be considered leaked.
*/
protected List<PooledConnection> getShallowCopy() {
ArrayList<PooledConnection> tmp = new ArrayList<PooledConnection>();
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.
*/
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java
+64 -40
View File
@@ -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.
* <p>
@@ -378,17 +402,19 @@ public class DataSourcePool implements DataSource {
* </p>
*/
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;
}
}
/**
src/main/java/com/avaje/ebeaninternal/server/lib/sql/FreeConnectionBuffer.java
+74 -147
View File
@@ -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.
* <p>
* It is circular in nature.
* </p>
* <p>
* All thread safety controlled externally (by PooledConnectionQueue).
* </p>
*
* @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<PooledConnection> freeBuffer = new LinkedList<PooledConnection>();
/**
* 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<PooledConnection> tempList = new ArrayList<PooledConnection>(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<PooledConnection> 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<PooledConnection> getShallowCopy() {
List<PooledConnection> copy = new ArrayList<PooledConnection>(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<PooledConnection> 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;
}
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java
+77 -19
View File
@@ -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.
* <p>
@@ -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<ExtendedPreparedStatement> 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.
* <p>
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java
+31 -71
View File
@@ -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<PooledConnection> 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<PooledConnection> 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();
}