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

Merge branch 'dbpool-logging'

Browse Source
This commit is contained in:
Rob Bygrave
2014-01-22 01:09:19 +13:00
parent acca803a80 e97e77e2e0
commit 87a948223c
18 changed files with 1401 additions and 807 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/avaje/ebean/config/DataSourceConfig.java
+69 -12
View File
@@ -36,25 +36,31 @@ public class DataSourceConfig {
private int heartbeatFreqSecs = 30;
private int heartbeatTimeoutSeconds = 3;
private boolean captureStackTrace;
private int maxStackTraceSize = 5;
private int leakTimeMinutes = 30;
private int maxInactiveTimeSecs = 900;
private int maxInactiveTimeSecs = 720;
private int maxAgeMinutes = 0;
private int trimPoolFreqSecs = 59;
private int pstmtCacheSize = 20;
private int cstmtCacheSize = 20;
private int waitTimeoutMillis = 1000;
private String poolListener;
private boolean offline;
Map<String, String> customProperties;
protected Map<String, String> customProperties;
/**
* Return the connection URL.
@@ -194,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
@@ -309,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.
@@ -321,6 +358,25 @@ public class DataSourceConfig {
this.maxInactiveTimeSecs = maxInactiveTimeSecs;
}
/**
* Return the minimum time gap between pool trim checks.
* <p>
* This defaults to 59 seconds meaning that the pool trim check will run every
* minute assuming the heart beat check runs every 30 seconds.
* </p>
*/
public int getTrimPoolFreqSecs() {
return trimPoolFreqSecs;
}
/**
* Set the minimum trim gap between pool trim checks.
*/
public void setTrimPoolFreqSecs(int trimPoolFreqSecs) {
this.trimPoolFreqSecs = trimPoolFreqSecs;
}
/**
* Return the pool listener.
*/
@@ -359,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.
*/
@@ -388,18 +444,18 @@ public class DataSourceConfig {
this.username = properties.get(prefix + "username", null);
this.password = properties.get(prefix + "password", null);
String v;
String dbDriver = properties.get(prefix + "databaseDriver", null);
this.driver = properties.get(prefix + "driver", dbDriver);
v = properties.get(prefix + "databaseDriver", null);
this.driver = properties.get(prefix + "driver", v);
v = properties.get(prefix + "databaseUrl", null);
this.url = properties.get(prefix + "url", v);
String dbUrl = properties.get(prefix + "databaseUrl", null);
this.url = properties.get(prefix + "url", dbUrl);
this.captureStackTrace = properties.getBoolean(prefix + "captureStackTrace", false);
this.maxStackTraceSize = properties.getInt(prefix + "maxStackTraceSize", 5);
this.leakTimeMinutes = properties.getInt(prefix + "leakTimeMinutes", 30);
this.maxInactiveTimeSecs = properties.getInt(prefix + "maxInactiveTimeSecs", 900);
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);
@@ -409,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
+5 -22
View File
@@ -353,6 +353,11 @@ public class DefaultServerFactory implements BootupEbeanManager {
if (sequenceFormat != null) {
nc.setSequenceFormat(sequenceFormat);
}
String schema = config.getProperty("namingConvention.schema");
if (schema != null) {
nc.setSchema(schema);
}
}
}
@@ -410,32 +415,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
+86 -20
View File
@@ -1,9 +1,11 @@
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;
import com.avaje.ebeaninternal.server.lib.sql.PooledConnectionStatistics.LoadValues;
/**
* A buffer especially designed for Busy PooledConnections.
@@ -21,6 +23,8 @@ import java.util.List;
*/
class BusyConnectionBuffer {
private static final Logger logger = LoggerFactory.getLogger(BusyConnectionBuffer.class);
private PooledConnection[] slots;
private int growBy;
@@ -46,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];
@@ -66,8 +70,8 @@ class BusyConnectionBuffer {
return size;
}
protected boolean isEmpty(){
return size == 0;
protected boolean isEmpty() {
return size == 0;
}
protected int add(PooledConnection pc){
@@ -75,41 +79,103 @@ class BusyConnectionBuffer {
// grow the capacity
setCapacity(slots.length + growBy);
}
++size;
int slot = nextEmptySlot();
pc.setSlotId(slot);
slots[slot] = pc;
return size;
return ++size;
}
protected boolean remove(PooledConnection pc) {
--size;
int slotId = pc.getSlotId();
if (slots[slotId] != pc){
PooledConnection heldBy = slots[slotId];
logger.warn("Failed to remove from slot[{}] PooledConnection[{}] - HeldBy[{}]", pc.getSlotId(), pc, heldBy);
return false;
}
slots[slotId] = null;
--size;
return true;
}
/**
* Collect the load statistics from all the busy connections.
* @param reset
*/
protected void collectStatistics(LoadValues values, boolean reset) {
for (int i = 0; i < slots.length; i++) {
if (slots[i] != null){
values.plus(slots[i].getStatistics().getValues(reset));
}
}
}
/**
* 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
+88 -44
View File
@@ -15,11 +15,12 @@ import java.util.Set;
import javax.persistence.PersistenceException;
import javax.sql.DataSource;
import com.avaje.ebean.config.DataSourceConfig;
import com.avaje.ebeaninternal.api.ClassUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.avaje.ebean.config.DataSourceConfig;
import com.avaje.ebeaninternal.api.ClassUtil;
/**
* A robust DataSource.
* <p>
@@ -73,8 +74,13 @@ public class DataSourcePool implements DataSource {
* The sql used to test a connection.
*/
private final String heartbeatsql;
private final int heartbeatFreqSecs;
private final int heartbeatTimeoutSeconds;
private final long trimPoolFreqMillis;
/**
* The transaction isolation level as per java.sql.Connection.
@@ -86,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).
*/
@@ -135,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;
@@ -166,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();
@@ -180,6 +193,8 @@ public class DataSourcePool implements DataSource {
this.waitTimeoutMillis = params.getWaitTimeoutMillis();
this.heartbeatsql = params.getHeartbeatSql();
this.heartbeatFreqSecs = params.getHeartbeatFreqSecs();
this.heartbeatTimeoutSeconds = params.getHeartbeatTimeoutSeconds();
this.trimPoolFreqMillis = 1000 * params.getTrimPoolFreqSecs();
queue = new PooledConnectionQueue(this);
@@ -317,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();
@@ -366,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>
@@ -374,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 + (maxInactiveTimeSecs * 1000))) {
queue.trim(maxInactiveTimeSecs);
lastTrimTime = System.currentTimeMillis();
if (testConnection(conn)) {
notifyDataSourceIsUp();
} else {
notifyDataSourceIsDown(null);
}
} catch (SQLException ex) {
@@ -487,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 {
@@ -539,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() + "]");
@@ -573,7 +602,15 @@ public class DataSourcePool implements DataSource {
}
queue.returnPooledConnection(pooledConnection);
}
/**
* Collect statistics of a connection that is fully closing
*/
protected void reportClosingConnection(PooledConnection pooledConnection) {
queue.reportClosingConnection(pooledConnection);
}
/**
* Returns information describing connections that are currently being used.
*/
@@ -635,7 +672,6 @@ public class DataSourcePool implements DataSource {
notifyDataSourceIsDown(ex);
throw ex;
}
}
/**
@@ -834,6 +870,14 @@ public class DataSourcePool implements DataSource {
public Status getStatus(boolean reset) {
return queue.getStatus(reset);
}
/**
* Return the aggregated load statistics collected on all the connections in the pool.
*/
public DataSourcePoolStatistics getStatistics(boolean reset) {
return queue.getStatistics(reset);
}
/**
* Deregister the JDBC driver.
@@ -873,8 +917,8 @@ public class DataSourcePool implements DataSource {
}
public String toString() {
return "min:" + minSize + " max:" + maxSize + " free:" + free + " busy:" + busy + " waiting:" + waiting
+ " highWaterMark:" + highWaterMark + " waitCount:" + waitCount + " hitCount:" + hitCount;
return "min[" + minSize + "] max[" + maxSize + "] free[" + free + "] busy[" + busy + "] waiting[" + waiting
+ "] highWaterMark[" + highWaterMark + "] waitCount[" + waitCount + "] hitCount[" + hitCount+"]";
}
/**
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePoolStatistics.java
+97
View File
@@ -0,0 +1,97 @@
package com.avaje.ebeaninternal.server.lib.sql;
/**
* Represents aggregated statistics collected from the DataSourcePool.
* <p>
* The goal is to present insight into the overload load of the DataSourcePool.
* These statistics can be collected and reported regularly to show load over
* time.
* </p>
* <p>
* Each pooled connection collects statistics. When a pooled connection is fully
* closed it can report it's statistics to the pool to be included as part of
* the collected statistics.
* </p>
*/
public class DataSourcePoolStatistics {
private final long collectionStart;
private final long count;
private final long errorCount;
private final long hwmMicros;
private final long totalMicros;
/**
* No statistics collected.
*/
public DataSourcePoolStatistics() {
this.collectionStart = 0;
this.count = 0;
this.errorCount = 0;
this.hwmMicros = 0;
this.totalMicros = 0;
}
/**
* Construct with statistics collected.
*/
public DataSourcePoolStatistics(long collectionStart, long count, long errorCount, long hwmMicros, long totalMicros) {
this.collectionStart = collectionStart;
this.count = count;
this.errorCount = errorCount;
this.hwmMicros = hwmMicros;
this.totalMicros = totalMicros;
}
public String toString() {
return "count[" + count + "] errors[" + errorCount + "] totalMicros[" + totalMicros + "] hwmMicros[" + hwmMicros
+ "] avgMicros[" + getAvgMicros() + "]";
}
/**
* Return the start time this set of statistics was collected from.
*/
public long getCollectionStart() {
return collectionStart;
}
/**
* Return the total number of 'get connection' requests.
*/
public long getCount() {
return count;
}
/**
* Return the number of SQLExceptions reported.
*/
public long getErrorCount() {
return errorCount;
}
/**
* Return the high water mark for the duration a connection was busy/used.
*/
public long getHwmMicros() {
return hwmMicros;
}
/**
* Return the aggregate time connections were busy/used.
*/
public long getTotalMicros() {
return totalMicros;
}
/**
* Return the average time connections were busy/used.
*/
public long getAvgMicros() {
return (totalMicros == 0) ? 0 : totalMicros / count;
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/ExtendedPreparedStatement.java
+3 -4
View File
@@ -31,18 +31,17 @@ public class ExtendedPreparedStatement extends ExtendedStatement implements Prep
/**
* The SQL used to create the underlying PreparedStatement.
*/
final String sql;
private final String sql;
/**
* The key used to cache this in the connection.
*/
final String cacheKey;
private final String cacheKey;
/**
* Create a wrapped PreparedStatement that can be cached.
*/
public ExtendedPreparedStatement(PooledConnection pooledConnection, PreparedStatement pstmt,
String sql, String cacheKey) {
public ExtendedPreparedStatement(PooledConnection pooledConnection, PreparedStatement pstmt, String sql, String cacheKey) {
super(pooledConnection, pstmt);
this.sql = sql;
this.cacheKey = cacheKey;
src/main/java/com/avaje/ebeaninternal/server/lib/sql/FreeConnectionBuffer.java
+79 -111
View File
@@ -1,138 +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 {
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>();
private int removeIndex;
protected FreeConnectionBuffer() {
}
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;
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);
}
protected boolean isEmpty() {
return size == 0;
}
/**
* Add at connection.
*/
protected void add(PooledConnection pc) {
conns[addIndex] = pc;
addIndex = inc(addIndex);
++size;
}
/**
* Trim any inactive connections that have not been used since usedSince.
*/
protected int trim(long usedSince, long createdSince) {
int trimCount = 0;
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;
}
/**
* Return a shallow copy of the free connections.
*/
protected 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;
}
/**
* Set the free list to be the connections in this copy. This is done after
* unused connections have been trimmed.
* <p>
* Not a particularly performant approach but this should not be called very
* often
* </p>
*/
protected void setShallowCopy(List<PooledConnection> copy) {
// reset to empty state
this.removeIndex = 0;
this.addIndex = 0;
this.size = 0;
return trimCount;
}
// null all the current connections
for (int i = 0; i < conns.length; i++) {
conns[i] = null;
}
/**
* Collect the load statistics from all the free connections.
*/
protected void collectStatistics(LoadValues values, boolean reset) {
// add connections 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));
}
/**
* 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;
this.conns = new PooledConnection[newCapacity];
// add the connections back from the copy
for (int i = 0; i < copy.size(); i++) {
add(copy.get(i));
}
}
}
/**
* 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
+488 -414
View File
File diff suppressed because it is too large Load Diff
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java
+86 -140
View File
@@ -1,18 +1,16 @@
package com.avaje.ebeaninternal.server.lib.sql;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import com.avaje.ebeaninternal.server.lib.sql.DataSourcePool.Status;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.avaje.ebeaninternal.server.lib.sql.DataSourcePool.Status;
import com.avaje.ebeaninternal.server.lib.sql.PooledConnectionStatistics.LoadValues;
public class PooledConnectionQueue {
private static final Logger logger = LoggerFactory.getLogger(PooledConnectionQueue.class);
@@ -34,6 +32,16 @@ public class PooledConnectionQueue {
*/
private final BusyConnectionBuffer busyList;
/**
* Load statistics collected off connections that have closed fully (left the pool).
*/
private final PooledConnectionStatistics collectedStats = new PooledConnectionStatistics();
/**
* Currently accumulated load statistics.
*/
private LoadValues accumulatedValues = new LoadValues();
/**
* Main lock guarding all access
*/
@@ -46,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;
@@ -94,11 +104,12 @@ public class PooledConnectionQueue {
this.warningSize = pool.getWarningSize();
this.waitTimeoutMillis = pool.getWaitTimeoutMillis();
this.leakTimeMinutes = pool.getLeakTimeMinutes();
this.busyList = new BusyConnectionBuffer(50,20);
this.freeList = new FreeConnectionBuffer(maxSize);
this.lock = new ReentrantLock(true);
this.maxAgeMillis = pool.getMaxAgeMillis();
this.busyList = new BusyConnectionBuffer(maxSize, 20);
this.freeList = new FreeConnectionBuffer();
this.lock = new ReentrantLock(false);
this.notEmpty = lock.newCondition();
}
@@ -116,6 +127,36 @@ public class PooledConnectionQueue {
}
}
/**
* Collect statistics of a connection that is fully closing
*/
protected void reportClosingConnection(PooledConnection pooledConnection) {
collectedStats.add(pooledConnection.getStatistics());
}
public DataSourcePoolStatistics getStatistics(boolean reset) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
LoadValues aggregate = collectedStats.getValues(reset);
freeList.collectStatistics(aggregate, reset);
busyList.collectStatistics(aggregate, reset);
aggregate.plus(accumulatedValues);
this.accumulatedValues = (reset) ? new LoadValues() : aggregate;
return new DataSourcePoolStatistics(aggregate.getCollectionStart(), aggregate.getCount(), aggregate.getErrorCount(), aggregate.getHwmMicros(), aggregate.getTotalMicros());
} finally {
lock.unlock();
}
}
public Status getStatus(boolean reset) {
final ReentrantLock lock = this.lock;
lock.lock();
@@ -152,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();
@@ -203,10 +244,11 @@ public class PooledConnectionQueue {
lock.lock();
try {
if (!busyList.remove(c)) {
logger.error("Connection [" + c + "] not found in BusyList? ");
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();
@@ -261,8 +303,7 @@ public class PooledConnectionQueue {
// are other threads already waiting? (they get priority)
if (waitingThreads == 0){
int freeSize = freeList.size();
if (freeSize > 0){
if (!freeList.isEmpty()){
// we have a free connection to return
return extractFromFreeList();
}
@@ -272,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;
}
@@ -333,13 +374,13 @@ public class PooledConnectionQueue {
try {
doingShutdown = true;
Status status = createStatus();
logger.debug("DataSourcePool [" + name + "] shutdown: "+status);
DataSourcePoolStatistics statistics = pool.getStatistics(false);
logger.debug("DataSourcePool [{}] shutdown {} - Statistics {}", name, status, statistics);
closeFreeConnections(true);
if (!busyList.isEmpty()) {
logger.warn("A potential connection leak was detected. Busy connections: "+ busyList.size());
logger.warn("Closing busy connections on shutdown size: "+ busyList.size());
dumpBusyConnectionInformation();
closeBusyConnections(0);
}
@@ -360,27 +401,30 @@ 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);
closeBusyConnections(leakTimeMinutes);
String busyMsg = "Busy Connections:\r\n" + getBusyConnectionInformation();
logger.info(busyMsg);
logger.info("Busy Connections:\n" + getBusyConnectionInformation());
} finally {
lock.unlock();
}
}
public void trim(int maxInactiveTimeSecs) throws SQLException {
public void trim(long maxInactiveMillis, long maxAgeMillis) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
trimInactiveConnections(maxInactiveTimeSecs);
ensureMinimumConnections();
if (trimInactiveConnections(maxInactiveMillis, maxAgeMillis) > 0) {
try {
ensureMinimumConnections();
} catch (SQLException e) {
logger.error("Error trying to ensure minimum connections", e);
}
}
} finally {
lock.unlock();
}
@@ -389,40 +433,14 @@ public class PooledConnectionQueue {
/**
* Trim connections that have been not used for some time.
*/
private int trimInactiveConnections(int maxInactiveTimeSecs) {
int maxTrim = freeList.size() - minSize;
if (maxTrim <= 0) {
return 0;
}
int trimedCount = 0;
long usedSince = System.currentTimeMillis() - (maxInactiveTimeSecs * 1000);
// get a shallow copy to manipulate
List<PooledConnection> freeListCopy = freeList.getShallowCopy();
Iterator<PooledConnection> it = freeListCopy.iterator();
while (it.hasNext()) {
PooledConnection pc = it.next();
if (pc.getLastUsedTime() < usedSince) {
// trim this connection as it hasn't been used in a while
trimedCount++;
it.remove();
pc.closeConnectionFully(true);
if (trimedCount >= maxTrim) {
break;
}
}
}
if (trimedCount > 0) {
private int trimInactiveConnections(long maxInactiveMillis, long maxAgeMillis) {
// rebuild the free list from the trimmed copy
freeList.setShallowCopy(freeListCopy);
String msg = "DataSourcePool [" + name + "] trimmed [" + trimedCount + "] inactive connections. New size[" + totalConnections() + "]";
logger.debug(msg);
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());
}
return trimedCount;
}
@@ -434,11 +452,7 @@ public class PooledConnectionQueue {
final ReentrantLock lock = this.lock;
lock.lock();
try {
while (!freeList.isEmpty()) {
PooledConnection c = freeList.remove();
logger.debug("PSTMT Statistics: "+c.getStatistics());
c.closeConnectionFully(logErrors);
}
freeList.closeAll(logErrors);
} finally {
lock.unlock();
}
@@ -461,62 +475,12 @@ public class PooledConnectionQueue {
final ReentrantLock lock = this.lock;
lock.lock();
try {
long olderThanTime = System.currentTimeMillis() - (leakTimeMinutes*60000);
List<PooledConnection> copy = busyList.getShallowCopy();
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);
}
}
busyList.closeBusyConnections(leakTimeMinutes);
} finally {
lock.unlock();
}
}
private void closeBusyConnection(PooledConnection pc) {
try {
String methodLine = pc.getCreatedByMethod();
Date luDate = new Date();
luDate.setTime(pc.getLastUsedTime());
String msg = "DataSourcePool closing leaked connection? " + " name["
+ pc.getName() + "] lastUsed[" + luDate + "] createdBy[" + methodLine
+ "] lastStmt[" + pc.getLastStatement() + "]";
logger.warn(msg);
logStackElement(pc, "Possible Leaked Connection: ");
System.out.println("CLOSING BUSY CONNECTION ??? "+pc);
pc.close();
} catch (SQLException ex) {
// this should never actually happen
logger.error(null, ex);
}
}
private void logStackElement(PooledConnection pc, String prefix) {
StackTraceElement[] stackTrace = pc.getStackTrace();
if (stackTrace != null){
String s = Arrays.toString(stackTrace);
String msg = prefix+" name["+pc.getName()+"] stackTrace: "+s;
logger.warn(msg);
// also send to syserr ... as the loggers get turned
// off early in JVM shutdown
System.err.println(msg);
}
}
/**
* As the pool grows it gets closer to the maxConnections limit. We can send
* an Alert (or warning) as we get close to this limit and hence an
@@ -557,26 +521,8 @@ public class PooledConnectionQueue {
lock.lock();
try {
if (toLogger) {
logger.info("Dumping busy connections: (Use datasource.xxx.capturestacktrace=true ... to get stackTraces)");
}
StringBuilder sb = new StringBuilder();
List<PooledConnection> copy = busyList.getShallowCopy();
for (int i = 0; i < copy.size(); i++) {
PooledConnection pc = copy.get(i);
if (toLogger) {
logger.info(pc.getDescription());
logStackElement(pc, "Busy Connection: ");
} else {
sb.append(pc.getDescription()).append("\r\n");
}
}
return sb.toString();
return busyList.getBusyConnectionInformation(toLogger);
} finally {
lock.unlock();
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionStatistics.java
+162
View File
@@ -0,0 +1,162 @@
package com.avaje.ebeaninternal.server.lib.sql;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
* Collects load statistics for a PooledConnection.
*/
public class PooledConnectionStatistics {
private final AtomicLong count = new AtomicLong();
private final AtomicLong errorCount = new AtomicLong();
private final AtomicLong hwmNanos = new AtomicLong();
private final AtomicLong totalNanos = new AtomicLong();
private final AtomicLong collectionStart;
public PooledConnectionStatistics() {
this.collectionStart = new AtomicLong(System.currentTimeMillis());
}
/**
* Add statistics from another collector.
*/
public void add(PooledConnectionStatistics other) {
errorCount.addAndGet(other.getErrorCount());
totalNanos.addAndGet(other.totalNanos.get());
count.addAndGet(other.getCount());
final long otherHwm = other.hwmNanos.get();
if (otherHwm > hwmNanos.get()) {
hwmNanos.set(otherHwm);
}
}
/**
* Add some time duration to the statistics.
*/
public void add(long durationNanos, boolean hasError) {
// This will be done in pretty much single threaded fashion
// as the Connections generally are not shared across threads
if (hasError) {
errorCount.incrementAndGet();
}
count.incrementAndGet();
totalNanos.addAndGet(durationNanos);
if (durationNanos > hwmNanos.get()) {
hwmNanos.set(durationNanos);
}
}
public String toString() {
return "count["+count+"] errors["+errorCount+"] totalMicros["+getTotalMicros()+"] hwmMicros["+getHwmMicros()+"]";
}
public long getCollectionStart() {
return collectionStart.get();
}
public long getCount() {
return count.get();
}
public long getErrorCount() {
return errorCount.get();
}
public long getTotalMicros() {
return TimeUnit.MICROSECONDS.convert(totalNanos.get(), TimeUnit.NANOSECONDS);
}
public long getHwmMicros() {
return TimeUnit.MICROSECONDS.convert(hwmNanos.get(), TimeUnit.NANOSECONDS);
}
/**
* Get the current values and reset the statistics if necessary.
*/
public LoadValues getValues(boolean reset) {
LoadValues value = new LoadValues(collectionStart.get(), count.get(), errorCount.get(), getHwmMicros(), getTotalMicros());
if (reset) {
count.set(0);
errorCount.set(0);
hwmNanos.set(0);
totalNanos.set(0);
collectionStart.set(System.currentTimeMillis());
}
return value;
}
/**
* Values representing the load or activity of a PooledConnection.
* <p>
* These are aggregated up to get a total for the DataSourcePool.
* </p>
*/
public static class LoadValues {
private long collectionStart;
private long count;
private long errorCount;
private long hwmMicros;
private long totalMicros;
public LoadValues() {
}
public LoadValues(long collectionStart, long count, long errorCount, long hwmMicros, long totalMicros) {
this.collectionStart = collectionStart;
this.count = count;
this.errorCount = errorCount;
this.hwmMicros = hwmMicros;
this.totalMicros = totalMicros;
}
public void plus(LoadValues additional) {
collectionStart = (collectionStart == 0) ? additional.collectionStart : Math.min(collectionStart, additional.collectionStart);
count += additional.count;
errorCount += additional.errorCount;
hwmMicros = Math.max(hwmMicros, additional.hwmMicros);
totalMicros += additional.totalMicros;
}
public String toString() {
return "count["+count+"] errors["+errorCount+"] totalMicros["+totalMicros+"] hwmMicros["+hwmMicros+"] avgMicros["+getAvgMicros()+"]";
}
public long getCollectionStart() {
return collectionStart;
}
public long getCount() {
return count;
}
public long getErrorCount() {
return errorCount;
}
public long getHwmMicros() {
return hwmMicros;
}
public long getTotalMicros() {
return totalMicros;
}
public long getAvgMicros() {
return (count == 0) ? 0 : totalMicros/count;
}
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PstmtCache.java
+26 -8
View File
@@ -14,37 +14,37 @@ public class PstmtCache extends LinkedHashMap<String, ExtendedPreparedStatement>
private static final Logger logger = LoggerFactory.getLogger(PstmtCache.class);
static final long serialVersionUID = -3096406924865550697L;
static final long serialVersionUID = -3096406924865550697L;
/**
* The name of the cache, for tracing purposes.
*/
final String cacheName;
protected final String cacheName;
/**
* The maximum size of the cache. When this is exceeded the oldest entry is removed.
*/
final int maxSize;
private final int maxSize;
/**
* The total number of entries removed from this cache.
*/
int removeCounter;
private int removeCounter;
/**
* The number of get hits.
*/
int hitCounter;
private int hitCounter;
/**
* The number of get() misses.
*/
int missCounter;
private int missCounter;
/**
* The number of puts into this cache.
*/
int putCounter;
private int putCounter;
public PstmtCache(String cacheName, int maxCacheSize) {
@@ -58,7 +58,7 @@ public class PstmtCache extends LinkedHashMap<String, ExtendedPreparedStatement>
* Return a summary description of this cache.
*/
public String getDescription() {
return cacheName+" size:"+size()+" max:"+maxSize+" totalHits:"+hitCounter+" hitRatio:"+getHitRatio()+" removes:"+removeCounter;
return "size["+size()+"] max["+maxSize+"] hits["+hitCounter+"] miss["+missCounter+"] hitRatio["+getHitRatio()+"] removes["+removeCounter+"]";
}
/**
@@ -101,6 +101,24 @@ public class PstmtCache extends LinkedHashMap<String, ExtendedPreparedStatement>
return putCounter;
}
/**
* Try to add the returning statement to the cache. If there is already a
* matching ExtendedPreparedStatement in the cache return false else add
* the statement to the cache and return true.
*/
public boolean returnStatement(ExtendedPreparedStatement pstmt) {
ExtendedPreparedStatement alreadyInCache = super.get(pstmt.getCacheKey());
if (alreadyInCache != null) {
return false;
}
// add the returning prepared statement to the cache.
// Note that the LRUCache will automatically close fully old unused
// PStmts when the cache has hit its maximum size.
put(pstmt.getCacheKey(), pstmt);
return true;
}
/**
* additionally maintains hit and miss statistics.
*/