Merge pull request #631 from wenshao/master

bug fixed for fatal exception & async close
2023-03-04 19:17:16 +00:00 · 2014-08-28 23:06:22 +08:00
parent 8e53ddceca 6caf8fe300
commit 2e03328bcb
5 changed files with 204 additions and 31 deletions
@@ -18,10 +18,12 @@ package com.alibaba.druid.pool;
 import com.alibaba.druid.support.logging.Log;
 import com.alibaba.druid.support.logging.LogFactory;
 import com.alibaba.druid.util.JdbcConstants;
+import com.alibaba.druid.util.JdbcUtils;
 import com.alibaba.druid.util.Utils;

 import javax.sql.ConnectionEventListener;
 import javax.sql.StatementEventListener;
+
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
@@ -232,7 +234,7 @@ public final class DruidConnectionHolder {

        for (Object item : statementTrace.toArray()) {
            Statement stmt = (Statement) item;
-            stmt.close();
+            JdbcUtils.close(stmt);
        }
        statementTrace.clear();

@@ -128,7 +128,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    // threads
    private ScheduledFuture<?>               destroySchedulerFuture;
    private DestroyTask                      destoryTask;
-    
+
    private CreateConnectionThread           createConnectionThread;
    private DestroyConnectionThread          destroyConnectionThread;
    private LogStatsThread                   logStatsThread;
@@ -155,7 +155,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    public static ThreadLocal<Long>          waitNanosLocal          = new ThreadLocal<Long>();

    private boolean                          logDifferentThread      = true;
-    
+
    public DruidDataSource(){
        this(false);
    }
@@ -527,7 +527,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                this.jdbcUrl = this.jdbcUrl.trim();
                initFromWrapDriverUrl();
            }
-            
+
            if (this.dbType == null || this.dbType.length() == 0) {
                this.dbType = JdbcUtils.getDbType(jdbcUrl, null);
            }
@@ -671,17 +671,18 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat

    protected void createAndStartDestroyThread() {
        destoryTask = new DestroyTask();
-        
+
        if (destroyScheduler != null) {
            long period = timeBetweenEvictionRunsMillis;
            if (period <= 0) {
                period = 1000;
            }
-            destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destoryTask, period, period, TimeUnit.MILLISECONDS);
+            destroySchedulerFuture = destroyScheduler.scheduleAtFixedRate(destoryTask, period, period,
+                                                                          TimeUnit.MILLISECONDS);
            initedLatch.countDown();
            return;
        }
-        
+
        String threadName = "Druid-ConnectionPool-Destroy-" + System.identityHashCode(this);
        destroyConnectionThread = new DestroyConnectionThread(threadName);
        destroyConnectionThread.start();
@@ -694,7 +695,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            createConnectionThread.start();
            return;
        }
-        
+
        initedLatch.countDown();
    }

@@ -883,6 +884,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    }

    private void initExceptionSorter() {
+        if (this.exceptionSorter != null) {
+            return;
+        }
+
        String realDriverClassName = driver.getClass().getName();
        if (realDriverClassName.equals(JdbcConstants.MYSQL_DRIVER)) {
            this.exceptionSorter = new MySqlExceptionSorter();
@@ -1152,10 +1157,20 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                        }
                    }
                }
-                this.discardConnection(holder.getConnection());
-                holder.setDiscard(true);
-                pooledConnection.disable(t);

+                boolean requireDiscard = false;
+                synchronized (pooledConnection) {
+                    if ((!pooledConnection.isClosed()) || !pooledConnection.isDisable()) {
+                        holder.setDiscard(true);
+                        pooledConnection.disable(t);
+                        requireDiscard = true;
+                    }
+                }
+                
+                if (requireDiscard) {
+                    this.discardConnection(holder.getConnection());
+                }
+                
                LOG.error("discard connection", sqlEx);
            }

@@ -1177,9 +1192,9 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        }

        if (logDifferentThread //
-                && (!isAsyncCloseConnectionEnable()) //
-                && pooledConnection.getOwnerThread() != Thread.currentThread()//
-                ) {
+            && (!isAsyncCloseConnectionEnable()) //
+            && pooledConnection.getOwnerThread() != Thread.currentThread()//
+        ) {
            LOG.warn("get/close not same thread");
        }

@@ -1300,7 +1315,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            if (destroyConnectionThread != null) {
                destroyConnectionThread.interrupt();
            }
-            
+
            if (destroySchedulerFuture != null) {
                destroySchedulerFuture.cancel(true);
            }
@@ -1479,11 +1494,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            return last;
        }
    }
-    
+
    private final void decrementPoolingCount() {
        poolingCount--;
    }
-    
+
    private final void incrementPoolingCount() {
        poolingCount++;
    }
@@ -1670,7 +1685,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    public long getRemoveAbandonedCount() {
        return removeAbandonedCount;
    }
-    
+
    protected void put(Connection connection) {
        DruidConnectionHolder holder = null;
        try {
@@ -1692,10 +1707,10 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat

            notEmpty.signal();
            notEmptySignalCount++;
-            
+
            if (createScheduler != null) {
                createTaskCount--;
-                
+
                if (poolingCount + createTaskCount < notEmptyWaitThreadCount //
                    && activeCount + poolingCount + createTaskCount < maxActive) {
                    emptySignal();
@@ -1705,10 +1720,11 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            lock.unlock();
        }
    }
-    
+
    public class CreateConnectionTask implements Runnable {
+
        private int errorCount = 0;
-        
+
        @Override
        public void run() {
            for (;;) {
@@ -1843,7 +1859,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                }

                put(connection);
-                
+
                errorCount = 0; // reset errorCount
            }
        }
@@ -1884,7 +1900,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
        }

    }
-    
+
    public class DestroyTask implements Runnable {

        @Override
@@ -1895,7 +1911,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
                removeAbandoned();
            }
        }
-        
+
    }

    public class LogStatsThread extends Thread {
@@ -2556,7 +2572,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
    public void setLogDifferentThread(boolean logDifferentThread) {
        this.logDifferentThread = logDifferentThread;
    }
-    
+
    public DruidPooledConnection tryGetConnection() throws SQLException {
        if (poolingCount == 0) {
            return null;
@@ -2646,7 +2662,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            return true;
        }
    }
-    
+
    public boolean isFull() {
        lock.lock();
        try {
@@ -2655,21 +2671,21 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat
            lock.unlock();
        }
    }
-    
+
    private void emptySignal() {
        if (createScheduler == null) {
            empty.signal();
            return;
        }
-        
+
        if (createTaskCount >= maxCreateTaskCount) {
            return;
        }
-        
+
        if (activeCount + poolingCount + createTaskCount >= maxActive) {
            return;
        }
-        
+
        createTaskCount++;
        CreateConnectionTask task = new CreateConnectionTask();
        createScheduler.submit(task);
@@ -40,6 +40,10 @@ public class Log4jImpl implements Log {
    public Log4jImpl(String loggerName){
        log = Logger.getLogger(loggerName);
    }
+    
+    public Logger getLog() {
+        return log;
+    }

    public boolean isDebugEnabled() {
        return log.isDebugEnabled();
@@ -32,6 +32,10 @@ public class AsyncCloseTest extends TestCase {
        closeExecutor = Executors.newFixedThreadPool(100);

    }
+    
+    protected void tearDown() throws Exception {
+        dataSource.close();
+    }

    public void test_0() throws Exception {
        for (int i = 0; i < 16; ++i) {
@@ -0,0 +1,147 @@
+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.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 AsyncCloseTest2 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(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.");
+        }
+    }
+
+    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);
+
+        Runnable connTask = new Runnable() {
+
+            @Override
+            public void run() {
+                try {
+                    Connection 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();
+                }
+            }
+        };
+
+        for (int i = 0; i < COUNT; ++i) {
+            connExecutor.submit(connTask);
+        }
+
+        closeLatch.await();
+        Assert.assertEquals(0, dataSource.getActiveCount());
+    }
+}