From 4ade284feed14ce3d33116ae4bd07d7afc00e9d2 Mon Sep 17 00:00:00 2001 From: wenshao Date: Fri, 22 Aug 2014 11:21:44 +0800 Subject: [PATCH] bug fixed for create/destroy scheduler & add testcase --- .../druid/pool/DruidAbstractDataSource.java | 15 +++ .../alibaba/druid/pool/DruidDataSource.java | 44 ++++++-- .../druid/bvt/pool/Large10ConcurrentTest.java | 103 ++++++++++++++++++ .../alibaba/druid/bvt/pool/Large10KTest.java | 52 +++++++++ 4 files changed, 204 insertions(+), 10 deletions(-) create mode 100644 src/test/java/com/alibaba/druid/bvt/pool/Large10ConcurrentTest.java create mode 100644 src/test/java/com/alibaba/druid/bvt/pool/Large10KTest.java diff --git a/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java index 9afb3228ce..e9a225b596 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java @@ -240,6 +240,7 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements protected DruidDataSourceStatLogger statLogger = new DruidDataSourceStatLoggerImpl(); private boolean asyncCloseConnectionEnable = false; + protected int maxCreateTaskCount = 2; protected ScheduledExecutorService destroyScheduler; protected ScheduledExecutorService createScheduler; @@ -1727,4 +1728,18 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements public boolean isInited() { return this.inited; } + + + public int getMaxCreateTaskCount() { + return maxCreateTaskCount; + } + + + public void setMaxCreateTaskCount(int maxCreateTaskCount) { + if (maxCreateTaskCount < 1) { + throw new IllegalArgumentException(); + } + + this.maxCreateTaskCount = maxCreateTaskCount; + } } diff --git a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java index c38856db50..e140e023af 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java @@ -617,7 +617,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat for (int i = 0, size = getInitialSize(); i < size; ++i) { Connection conn = createPhysicalConnection(); DruidConnectionHolder holder = new DruidConnectionHolder(this, conn); - connections[poolingCount++] = holder; + connections[poolingCount] = holder; + incrementPoolingCount(); } if (poolingCount > 0) { @@ -1175,7 +1176,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat return; } - if (logDifferentThread && pooledConnection.getOwnerThread() != Thread.currentThread()) { + if (logDifferentThread // + && (!isAsyncCloseConnectionEnable()) // + && pooledConnection.getOwnerThread() != Thread.currentThread()// + ) { LOG.warn("get/close not same thread"); } @@ -1376,7 +1380,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat void putLast(DruidConnectionHolder e, long lastActiveTimeMillis) { e.setLastActiveTimeMillis(lastActiveTimeMillis); - connections[poolingCount++] = e; + connections[poolingCount] = e; + incrementPoolingCount(); if (poolingCount > poolingPeak) { poolingPeak = poolingCount; @@ -1413,7 +1418,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat throw ie; } - poolingCount--; + decrementPoolingCount(); DruidConnectionHolder last = connections[poolingCount]; connections[poolingCount] = null; @@ -1467,13 +1472,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } } - poolingCount--; + decrementPoolingCount(); DruidConnectionHolder last = connections[poolingCount]; connections[poolingCount] = null; return last; } } + + private final void decrementPoolingCount() { + poolingCount--; + } + + private final void incrementPoolingCount() { + poolingCount++; + } @Override public Connection getConnection(String username, String password) throws SQLException { @@ -1669,7 +1682,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat lock.lock(); try { - connections[poolingCount++] = holder; + connections[poolingCount] = holder; + incrementPoolingCount(); if (poolingCount > poolingPeak) { poolingPeak = poolingCount; @@ -1681,6 +1695,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (createScheduler != null) { createTaskCount--; + + if (poolingCount + createTaskCount < notEmptyWaitThreadCount // + && activeCount + poolingCount + createTaskCount < maxActive) { + emptySignal(); + } } } finally { lock.unlock(); @@ -1727,7 +1746,8 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (errorCount > connectionErrorRetryAttempts && timeBetweenConnectErrorMillis > 0) { if (breakAfterAcquireFailure) { - break; + createTaskCount--; + return; } createScheduler.schedule(this, timeBetweenConnectErrorMillis, TimeUnit.MILLISECONDS); @@ -1746,6 +1766,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } put(connection); + break; } } } @@ -2641,13 +2662,16 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat return; } - if (createTaskCount >= 2) { + if (createTaskCount >= maxCreateTaskCount) { return; } - CreateConnectionTask task = new CreateConnectionTask(); - createScheduler.submit(task); + if (activeCount + poolingCount + createTaskCount >= maxActive) { + return; + } createTaskCount++; + CreateConnectionTask task = new CreateConnectionTask(); + createScheduler.submit(task); } } diff --git a/src/test/java/com/alibaba/druid/bvt/pool/Large10ConcurrentTest.java b/src/test/java/com/alibaba/druid/bvt/pool/Large10ConcurrentTest.java new file mode 100644 index 0000000000..ec62a12a7f --- /dev/null +++ b/src/test/java/com/alibaba/druid/bvt/pool/Large10ConcurrentTest.java @@ -0,0 +1,103 @@ +package com.alibaba.druid.bvt.pool; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicLong; + +import javax.sql.DataSource; + +import junit.framework.TestCase; + +import org.junit.Assert; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.util.JdbcUtils; + +public class Large10ConcurrentTest extends TestCase { + + private DruidDataSource[] dataSources = new DruidDataSource[10000]; + private ScheduledExecutorService scheduler; + + private ExecutorService executor; + + protected void setUp() throws Exception { + executor = Executors.newFixedThreadPool(100); + scheduler = Executors.newScheduledThreadPool(10); + for (int i = 0; i < dataSources.length; ++i) { + DruidDataSource dataSource = new DruidDataSource(); + dataSource.setMaxActive(8); + dataSource.setMinIdle(0); + dataSource.setUrl("jdbc:mock:xxx"); + dataSource.setCreateScheduler(scheduler); + dataSource.setDestroyScheduler(scheduler); + dataSource.setTestOnBorrow(false); + dataSource.setTestWhileIdle(false); + dataSource.setAsyncCloseConnectionEnable(true); + + dataSources[i] = dataSource; + } + } + + protected void tearDown() throws Exception { + for (int i = 0; i < dataSources.length; ++i) { + JdbcUtils.close(dataSources[i]); + } + executor.shutdown(); + scheduler.shutdown(); + } + + public void test_large() throws Exception { + final Connection[] connections = new Connection[dataSources.length * 8]; + final CountDownLatch connLatch = new CountDownLatch(connections.length); + final AtomicLong connErrorCount = new AtomicLong(); + + for (int i = 0; i < dataSources.length; ++i) { + for (int j = 0; j < 8; ++j) { + final DataSource dataSource = dataSources[i]; + final int index = i * 8 + j; + Runnable task = new Runnable() { + + public void run() { + try { + connections[index] = dataSource.getConnection(); + } catch (SQLException e) { + connErrorCount.incrementAndGet(); + e.printStackTrace(); + } finally { + connLatch.countDown(); + } + } + }; + executor.execute(task); + } + } + connLatch.await(); + + for (int i = 0; i < dataSources.length; ++i) { + Assert.assertEquals(8, dataSources[i].getActiveCount()); + } + + for (int i = 0; i < dataSources.length; ++i) { + Assert.assertEquals(0, dataSources[i].getPoolingCount()); + } + + final CountDownLatch closeLatch = new CountDownLatch(connections.length); + for (int i = 0; i < dataSources.length; ++i) { + for (int j = 0; j < 8; ++j) { + final int index = i * 8 + j; + Runnable task = new Runnable() { + public void run() { + JdbcUtils.close(connections[index]); + closeLatch.countDown(); + } + }; + executor.execute(task); + } + } + closeLatch.await(); + } +} diff --git a/src/test/java/com/alibaba/druid/bvt/pool/Large10KTest.java b/src/test/java/com/alibaba/druid/bvt/pool/Large10KTest.java new file mode 100644 index 0000000000..d07cafde97 --- /dev/null +++ b/src/test/java/com/alibaba/druid/bvt/pool/Large10KTest.java @@ -0,0 +1,52 @@ +package com.alibaba.druid.bvt.pool; + +import java.sql.Connection; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +import junit.framework.TestCase; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.util.JdbcUtils; + +public class Large10KTest extends TestCase { + + private DruidDataSource[] dataSources = new DruidDataSource[10000]; + private ScheduledExecutorService scheduler; + + protected void setUp() throws Exception { + scheduler = Executors.newScheduledThreadPool(10); + for (int i = 0; i < dataSources.length; ++i) { + DruidDataSource dataSource = new DruidDataSource(); + dataSource.setUrl("jdbc:mock:xxx"); + dataSource.setCreateScheduler(scheduler); + dataSource.setDestroyScheduler(scheduler); + dataSource.setTestOnBorrow(false); + dataSource.setTestWhileIdle(false); + + dataSources[i] = dataSource; + } + } + + protected void tearDown() throws Exception { + for (int i = 0; i < dataSources.length; ++i) { + JdbcUtils.close(dataSources[i]); + } + scheduler.shutdown(); + } + + public void test_large() throws Exception { + Connection[] connections = new Connection[dataSources.length * 8]; + for (int i = 0; i < dataSources.length; ++i) { + for (int j = 0; j < 8; ++j) { + connections[i * 8 + j] = dataSources[i].getConnection(); + } + } + + for (int i = 0; i < dataSources.length; ++i) { + for (int j = 0; j < 8; ++j) { + connections[i * 8 + j].close(); + } + } + } +}