diff --git a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java index 7994771916..eecc295a1b 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java @@ -1169,6 +1169,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat if (requireDiscard) { this.discardConnection(holder.getConnection()); + holder.setDiscard(true); } LOG.error("discard connection", sqlEx); @@ -1225,7 +1226,18 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat } // reset holder, restore default settings, clear warnings - holder.reset(); + boolean isSameThread = pooledConnection.getOwnerThread() == Thread.currentThread(); + if (!isSameThread) { + synchronized (pooledConnection) { + holder.reset(); + } + } else { + holder.reset(); + } + + if (holder.isDiscard()) { + return; + } if (testOnReturn) { boolean validate = testConnectionInternal(physicalConnection); diff --git a/src/main/java/com/alibaba/druid/pool/DruidPooledConnection.java b/src/main/java/com/alibaba/druid/pool/DruidPooledConnection.java index 4ca971e6f3..c4b831014a 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidPooledConnection.java +++ b/src/main/java/com/alibaba/druid/pool/DruidPooledConnection.java @@ -217,7 +217,14 @@ public class DruidPooledConnection extends PoolableWrapper implements javax.sql. return; } - if (holder.getDataSource().isAsyncCloseConnectionEnable()) { + DruidAbstractDataSource dataSource = holder.getDataSource(); + boolean isSameThread = this.getOwnerThread() == Thread.currentThread(); + + if (!isSameThread) { + dataSource.setAsyncCloseConnectionEnable(true); + } + + if (dataSource.isAsyncCloseConnectionEnable()) { syncClose(); return; } @@ -226,7 +233,7 @@ public class DruidPooledConnection extends PoolableWrapper implements javax.sql. listener.connectionClosed(new ConnectionEvent(this)); } - DruidAbstractDataSource dataSource = holder.getDataSource(); + List filters = dataSource.getProxyFilters(); if (filters.size() > 0) { FilterChainImpl filterChain = new FilterChainImpl(dataSource); diff --git a/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest1.java b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest1.java new file mode 100644 index 0000000000..600c9f765f --- /dev/null +++ b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest1.java @@ -0,0 +1,117 @@ +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.atomic.AtomicInteger; + +import junit.framework.TestCase; + +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.junit.Assert; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.support.logging.Log4jImpl; +import com.alibaba.druid.support.logging.LogFactory; + +public class AsyncCloseTest1 extends TestCase { + + protected DruidDataSource dataSource; + private ExecutorService connExecutor; + private ExecutorService closeExecutor; + + final AtomicInteger errorCount = new AtomicInteger(); + + private Logger log; + private Level oldLevel; + + protected void setUp() throws Exception { + log = ((Log4jImpl) LogFactory.getLog(DruidDataSource.class)).getLog(); + oldLevel = log.getLevel(); + log.setLevel(Level.FATAL); + + dataSource = new DruidDataSource(); + dataSource.setUrl("jdbc:mock:"); +// dataSource.setAsyncCloseConnectionEnable(true); + dataSource.setTestOnBorrow(false); + dataSource.setMaxActive(16); + + connExecutor = Executors.newFixedThreadPool(128); + closeExecutor = Executors.newFixedThreadPool(128); + + } + + protected void tearDown() throws Exception { + dataSource.close(); + log.setLevel(oldLevel); + } + + public void test_0() throws Exception { + for (int i = 0; i < 16; ++i) { + loop(); + System.out.println("loop " + i + " done."); + } + } + + class CloseTask implements Runnable { + + private Connection conn; + private CountDownLatch latch; + + public CloseTask(Connection conn, CountDownLatch latch){ + this.conn = conn; + this.latch = latch; + } + + @Override + public void run() { + try { + conn.close(); + } catch (SQLException e) { + errorCount.incrementAndGet(); + } finally { + latch.countDown(); + } + } + + } + + protected void loop() throws InterruptedException { + dataSource.shrink(); + Assert.assertEquals(0, dataSource.getActiveCount()); + Assert.assertEquals(0, dataSource.getPoolingCount()); + + final int COUNT = 1024 * 128; + final CountDownLatch closeLatch = new CountDownLatch(COUNT * 2); + + Runnable connTask = new Runnable() { + + @Override + public void run() { + try { + Connection conn = dataSource.getConnection(); + + CloseTask closeTask = new CloseTask(conn, closeLatch); + + closeExecutor.submit(closeTask); + closeExecutor.submit(closeTask); // dup close + } catch (SQLException e) { + errorCount.incrementAndGet(); + } + } + }; + + for (int i = 0; i < COUNT; ++i) { + connExecutor.submit(connTask); + } + + closeLatch.await(); + Assert.assertEquals(0, dataSource.getActiveCount()); + Assert.assertEquals(16, dataSource.getPoolingCount()); + + + } +} diff --git a/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest3.java b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest3.java new file mode 100644 index 0000000000..7b987cc580 --- /dev/null +++ b/src/test/java/com/alibaba/druid/bvt/pool/AsyncCloseTest3.java @@ -0,0 +1,152 @@ +package com.alibaba.druid.bvt.pool; + +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import junit.framework.TestCase; + +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.junit.Assert; + +import com.alibaba.druid.filter.FilterAdapter; +import com.alibaba.druid.filter.FilterChain; +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.pool.DruidPooledConnection; +import com.alibaba.druid.pool.vendor.MockExceptionSorter; +import com.alibaba.druid.proxy.jdbc.StatementProxy; +import com.alibaba.druid.support.logging.Log4jImpl; +import com.alibaba.druid.support.logging.LogFactory; + +public class AsyncCloseTest3 extends TestCase { + + protected DruidDataSource dataSource; + private ExecutorService connExecutor; + private ExecutorService closeExecutor; + + final AtomicInteger errorCount = new AtomicInteger(); + + private Logger log; + private Level oldLevel; + + protected void setUp() throws Exception { + log = ((Log4jImpl) LogFactory.getLog(DruidDataSource.class)).getLog(); + oldLevel = log.getLevel(); + log.setLevel(Level.FATAL); + + dataSource = new DruidDataSource(); + dataSource.setUrl("jdbc:mock:"); +// dataSource.setAsyncCloseConnectionEnable(true); + dataSource.setTestOnBorrow(false); + dataSource.setMaxActive(16); + + dataSource.getProxyFilters().add(new FilterAdapter() { + + @Override + public boolean statement_execute(FilterChain chain, StatementProxy statement, String sql) + throws SQLException { + throw new SQLException(); + } + }); + + dataSource.setExceptionSorter(new MyExceptionSorter()); + dataSource.init(); + + connExecutor = Executors.newFixedThreadPool(2); + closeExecutor = Executors.newFixedThreadPool(2); + + } + + protected void tearDown() throws Exception { + dataSource.close(); + log.setLevel(oldLevel); + } + + public void test_0() throws Exception { + for (int i = 0; i < 16; ++i) { + loop(); + System.out.println("loop " + i + " done."); + } + } + + public static class MyExceptionSorter extends MockExceptionSorter { + + @Override + public boolean isExceptionFatal(SQLException e) { + return true; + } + } + + class CloseTask implements Runnable { + + private Connection conn; + private CountDownLatch latch; + + public CloseTask(Connection conn, CountDownLatch latch){ + this.conn = conn; + this.latch = latch; + } + + @Override + public void run() { + try { + conn.close(); + } catch (SQLException e) { + errorCount.incrementAndGet(); + } finally { + latch.countDown(); + } + } + + } + + protected void loop() throws InterruptedException { + dataSource.shrink(); + Assert.assertEquals(0, dataSource.getActiveCount()); + Assert.assertEquals(0, dataSource.getPoolingCount()); + + final int COUNT = 1024 * 128; + final CountDownLatch closeLatch = new CountDownLatch(COUNT * 2); + final CountDownLatch execLatch = new CountDownLatch(COUNT); + + Runnable connTask = new Runnable() { + + @Override + public void run() { + try { + DruidPooledConnection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + + CloseTask closeTask = new CloseTask(conn, closeLatch); + + closeExecutor.submit(closeTask); + closeExecutor.submit(closeTask); // dup close + + try { + stmt.execute("select 1"); + } finally { + stmt.close(); + conn.close(); + } + } catch (SQLException e) { + errorCount.incrementAndGet(); + } finally { + execLatch.countDown(); + } + } + }; + + for (int i = 0; i < COUNT; ++i) { + connExecutor.submit(connTask); + } + + execLatch.await(); + closeLatch.await(); + Assert.assertEquals(0, dataSource.getActiveCount()); + } +} diff --git a/src/test/java/com/alibaba/druid/bvt/pool/exception/OracleExceptionSorterTest_closeConn.java b/src/test/java/com/alibaba/druid/bvt/pool/exception/OracleExceptionSorterTest_closeConn.java index eda5aa3b1c..ff8bb81f22 100644 --- a/src/test/java/com/alibaba/druid/bvt/pool/exception/OracleExceptionSorterTest_closeConn.java +++ b/src/test/java/com/alibaba/druid/bvt/pool/exception/OracleExceptionSorterTest_closeConn.java @@ -63,7 +63,9 @@ public class OracleExceptionSorterTest_closeConn extends TestCase { SQLException exception = new SQLException("xx", "xxx", 28); mockConn.setError(exception); + Assert.assertEquals(1, dataSource.getActiveCount()); conn.close(); + Assert.assertEquals(0, dataSource.getActiveCount()); { Connection conn2 = dataSource.getConnection();