diff --git a/pom.xml b/pom.xml index 03033ccac..7358781ca 100644 --- a/pom.xml +++ b/pom.xml @@ -140,6 +140,13 @@ test + + org.mockito + mockito-core + 1.9.5 + test + + ch.qos.logback logback-classic diff --git a/src/main/java/com/avaje/ebean/config/DataSourceConfig.java b/src/main/java/com/avaje/ebean/config/DataSourceConfig.java index 9ed13c363..e2bde6dbe 100644 --- a/src/main/java/com/avaje/ebean/config/DataSourceConfig.java +++ b/src/main/java/com/avaje/ebean/config/DataSourceConfig.java @@ -43,6 +43,8 @@ public class DataSourceConfig { private int leakTimeMinutes = 30; private int maxInactiveTimeSecs = 720; + + private int trimPoolFreqSecs = 59; private int pstmtCacheSize = 20; @@ -321,6 +323,25 @@ public class DataSourceConfig { this.maxInactiveTimeSecs = maxInactiveTimeSecs; } + + /** + * Return the minimum time gap between pool trim checks. + *

+ * This defaults to 59 seconds meaning that the pool trim check will run every + * minute assuming the heart beat check runs every 30 seconds. + *

+ */ + 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. */ @@ -388,18 +409,17 @@ 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.minConnections = properties.getInt(prefix + "minConnections", 0); this.maxConnections = properties.getInt(prefix + "maxConnections", 20); 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 3cb861427..0dff23448 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 @@ -77,6 +77,8 @@ public class DataSourcePool implements DataSource { private final int heartbeatFreqSecs; + private final int trimPoolFreqSecs; + /** * The transaction isolation level as per java.sql.Connection. */ @@ -181,6 +183,7 @@ public class DataSourcePool implements DataSource { this.waitTimeoutMillis = params.getWaitTimeoutMillis(); this.heartbeatsql = params.getHeartbeatSql(); this.heartbeatFreqSecs = params.getHeartbeatFreqSecs(); + this.trimPoolFreqSecs = params.getTrimPoolFreqSecs(); queue = new PooledConnectionQueue(this); @@ -383,7 +386,7 @@ public class DataSourcePool implements DataSource { notifyDataSourceIsUp(); - if (System.currentTimeMillis() > (lastTrimTime + (maxInactiveTimeSecs * 1000))) { + if (System.currentTimeMillis() > (lastTrimTime + (trimPoolFreqSecs * 1000))) { queue.trim(maxInactiveTimeSecs); lastTrimTime = System.currentTimeMillis(); } 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 813740633..6186aebd8 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 @@ -19,10 +19,19 @@ import com.avaje.ebeaninternal.server.lib.sql.PooledConnectionStatistics.LoadVal */ class FreeConnectionBuffer { + /** + * The buffer itself. + */ private PooledConnection[] conns; + /** + * The position in the buffer where the next connection is removed from. + */ private int removeIndex; + /** + * Position in the buffer where the next connection is added to. + */ private int addIndex; /** @@ -51,13 +60,16 @@ class FreeConnectionBuffer { */ protected void add(PooledConnection pc) { if (conns[addIndex] != null) { - throw new RuntimeException("Buffer slot ["+addIndex+"] already full?"); + throw new IllegalStateException("Buffer slot ["+addIndex+"] already full?"); } conns[addIndex] = pc; addIndex = inc(addIndex); ++size; } + /** + * Close all connections in this buffer. + */ protected void closeAll(boolean logErrors) { final PooledConnection[] items = this.conns; @@ -88,19 +100,25 @@ class FreeConnectionBuffer { } /** - * Return a shallow copy of the free connections. + * Trim any inactive connections that have not been used since usedSince. */ - protected List getShallowCopy() { - - List copy = new ArrayList(conns.length); - for (int i = 0; i < conns.length; i++) { - if (conns[i] != null){ - copy.add(conns[i]); - } + 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 copy; + } + + return trimCount; } - + /** * Collect the load statistics from all the free connections. */ @@ -114,36 +132,25 @@ class FreeConnectionBuffer { } /** - * Set the free list to be the connections in this copy. This is done after - * unused connections have been trimmed. - *

- * Not a particularly performant approach but this should not be called very - * often - *

+ * Return a shallow copy of the free connections. */ - protected void setShallowCopy(List copy) { - - // reset to empty state - this.removeIndex = 0; - this.addIndex = 0; - this.size = 0; - - // null all the current connections - for (int i = 0; i < conns.length; i++) { - conns[i] = null; - } - - // add connections from the copy - for (int i = 0; i < copy.size(); i++) { - add(copy.get(i)); - } - } + 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(); 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 d6c3293bc..9712d5482 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 @@ -9,10 +9,12 @@ import java.sql.SQLWarning; 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; @@ -218,10 +220,18 @@ public class PooledConnection extends ConnectionDelegator { return getDescription(); } - public String getDescription() { - return "name["+name+"] slot["+slotId+"] startTime["+getStartUseTime()+"] stmt["+getLastStatement()+"] createdBy["+getCreatedByMethod()+"]"; + public long getBusySeconds() { + return (System.currentTimeMillis() - startUseTime)/1000; } - + + public String getDescription() { + return "name["+name+"] slot["+slotId+"] startTime["+getStartUseTime()+"] busySeconds["+getBusySeconds()+"] createdBy["+getCreatedByMethod()+"] stmt["+getLastStatement()+"]"; + } + + public String getFullDescription() { + return "name["+name+"] slot["+slotId+"] startTime["+getStartUseTime()+"] busySeconds["+getBusySeconds()+"] stackTrace["+getStackTraceAsString()+"] stmt["+getLastStatement()+"]"; + } + public String getPstmtStatistics() { return "name["+name+"] startTime["+getStartUseTime()+"] "+pstmtCache.getDescription(); } @@ -907,6 +917,17 @@ public class PooledConnection extends ConnectionDelegator { this.stackTrace = stackTrace; } + /** + * Return the stackTrace as a String for logging purposes. + */ + public String getStackTraceAsString() { + StackTraceElement[] stackTrace = getStackTrace(); + if (stackTrace == null){ + return ""; + } + return Arrays.toString(stackTrace); + } + /** * Return the full stack trace that got the connection from the pool. You * could use this if getCreatedByMethod() doesn't work for you. 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 fe5d45c04..1076f9ad0 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,9 +1,6 @@ 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; @@ -418,9 +415,9 @@ public class PooledConnectionQueue { final ReentrantLock lock = this.lock; lock.lock(); try { - trimInactiveConnections(maxInactiveTimeSecs); - ensureMinimumConnections(); - + if (trimInactiveConnections(maxInactiveTimeSecs) > 0) { + ensureMinimumConnections(); + } } finally { lock.unlock(); } @@ -430,39 +427,12 @@ 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 freeListCopy = freeList.getShallowCopy(); - - Iterator 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; - } - } - } - + int trimedCount = freeList.trim(usedSince); if (trimedCount > 0) { - - // 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); + logger.debug("DataSourcePool [{}] trimmed [{}] inactive connections. New size[{}]", name, trimedCount, totalConnections()); } return trimedCount; } @@ -522,18 +492,9 @@ public class PooledConnectionQueue { private void closeBusyConnection(PooledConnection pc, long leakMinutes) { try { - String methodLine = pc.getCreatedByMethod(); - - Date luDate = new Date(); - luDate.setTime(pc.getLastUsedTime()); - - String msg = "DataSourcePool closing leaked connection? name[" - + pc.getName() + "] leakMinutes["+leakMinutes+"] lastUsed[" + luDate + "] createdBy[" + methodLine - + "] lastStmt[" + pc.getLastStatement() + "]"; - - logger.warn(msg); - logStackElement(pc, "Possible Leaked Connection: "); - System.out.println("CLOSING Possibly leaked connection: "+pc); + + logger.warn("DataSourcePool closing busy connection? "+pc.getFullDescription()); + System.out.println("CLOSING busy connection: "+pc.getFullDescription()); pc.closeConnectionFully(false); @@ -543,19 +504,6 @@ public class PooledConnectionQueue { } } - 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 @@ -607,11 +555,9 @@ public class PooledConnectionQueue { for (int i = 0; i < copy.size(); i++) { PooledConnection pc = copy.get(i); if (toLogger) { - logger.info(pc.getDescription()); - logStackElement(pc, "Busy Connection: "); - + logger.info("Busy Connection - {}", pc.getFullDescription()); } else { - sb.append(pc.getDescription()).append("\r\n"); + sb.append(pc.getFullDescription()).append("\r\n"); } } 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 cd3ee4b7f..26fe97c91 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 @@ -1,11 +1,13 @@ package com.avaje.ebeaninternal.server.lib.sql; import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; -import org.junit.Assert; 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; @@ -15,38 +17,42 @@ public class TestDataSourceMax extends BaseTestCase { @Test public void test() { - boolean runThisManuallyNow = true; + boolean skipThisTest = true; - if (!runThisManuallyNow) { + if (skipThisTest) { return; } + Ebean.getServer(null); + String name = "h2"; DataSourceConfig dsConfig = new DataSourceConfig(); dsConfig.loadSettings(name); - dsConfig.setMinConnections(3); - dsConfig.setMaxConnections(3); + dsConfig.setMinConnections(2); + dsConfig.setMaxConnections(8); dsConfig.setWaitTimeoutMillis(30000); + dsConfig.setCaptureStackTrace(true); DataSourcePool pool = new DataSourcePool(null, name, dsConfig); - Assert.assertEquals(3, pool.getMaxSize()); + // Assert.assertEquals(3, pool.getMaxSize()); DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(10, 2, 180, 30, "testDs"); try { for (int i = 0; i < 12; i++) { // Thread.sleep(10*i); - bg.execute(new ConnRunner(pool, 100)); + bg.execute(new ConnRunner(pool, 4000)); } System.out.println("main thread sleep ... " + pool.getStatus(false)); - Thread.sleep(1000); + Thread.sleep(10000); 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. @@ -71,14 +77,43 @@ public class TestDataSourceMax extends BaseTestCase { } public void run() { + Connection connection = null; + PreparedStatement pstmt = null; + ResultSet rset = null; try { - Connection connection = pool.getConnection(); + connection = pool.getConnection(); + pstmt = connection.prepareStatement("select count(*) from o_customer"); + rset = pstmt.executeQuery(); + + System.out.println("sleep " + sleepMillis); Thread.sleep(sleepMillis); - connection.close(); - } catch (Exception e) { - e.printStackTrace(); + System.out.println("sleep done"); + + } catch (Exception ex) { + ex.printStackTrace(); + } finally { + if (rset != null) { + try { + rset.close(); + } catch (Exception e) { + e.printStackTrace(); + } + if (pstmt != null) { + try { + pstmt.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + if (connection != null) { + try { + connection.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + } } } - } } 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 new file mode 100644 index 000000000..9a3c8b293 --- /dev/null +++ b/src/test/java/com/avaje/ebeaninternal/server/lib/sql/TestFreeBufferTrim.java @@ -0,0 +1,63 @@ +package com.avaje.ebeaninternal.server.lib.sql; + +import org.junit.Assert; +import org.junit.Test; +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); + Assert.assertEquals(0, b.size()); + + PooledConnection p0 = Mockito.mock(PooledConnection.class); + Mockito.when(p0.getLastUsedTime()).thenReturn(1000l); + + PooledConnection p1 = Mockito.mock(PooledConnection.class); + Mockito.when(p1.getLastUsedTime()).thenReturn(2000l); + + PooledConnection p2 = Mockito.mock(PooledConnection.class); + Mockito.when(p2.getLastUsedTime()).thenReturn(1100l); + + b.add(p0); + b.add(p1); + b.add(p2); + + Assert.assertEquals(3, b.size()); + + int trimCount = b.trim(1500); + + Assert.assertEquals(1, b.size()); + Assert.assertEquals(2, trimCount); + } + + + +}