Close connections asynchroniously

Author: rPraml

ebean-datasource/src/main/java/io/ebean/datasource/pool/BusyConnectionBuffer.java

@@ -107,7 +107,6 @@ void closeBusyConnections(long leakTimeMinutes) {
   private void closeBusyConnection(PooledConnection pc) {
     try {
       Log.warn("DataSource closing busy connection? {0}", pc.fullDescription());
-      System.out.println("CLOSING busy connection: " + pc.fullDescription());
       pc.closeConnectionFully(false);
     } catch (Exception ex) {
       Log.error("Error when closing potentially leaked connection " + pc.description(), ex);

ebean-datasource/src/main/java/io/ebean/datasource/pool/ConnectionPool.java

@@ -6,6 +6,11 @@
 import java.io.PrintWriter;
 import java.sql.*;
 import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.LongAdder;
@@ -26,7 +31,7 @@
 final class ConnectionPool implements DataSourcePool {
 
   private static final String APPLICATION_NAME = "ApplicationName";
-  private final ReentrantLock heartbeatLock = new ReentrantLock(false);
+  private final ReentrantLock backgroundtasksLock = new ReentrantLock(false);
   private final ReentrantLock notifyLock = new ReentrantLock(false);
   /**
    * The name given to this dataSource.
@@ -94,6 +99,8 @@ final class ConnectionPool implements DataSourcePool {
   private final boolean shutdownOnJvmExit;
   private Thread shutdownHook;
 
+  private ExecutorService executor;
+
   ConnectionPool(String name, DataSourceConfig params) {
     this.config = params;
     this.name = name;
@@ -192,7 +199,7 @@ private void initialiseConnections() throws SQLException {
     } else {
       tryEnsureMinimumConnections();
     }
-    startHeartBeatIfStopped();
+    startBackgroundActionsIfStopped();
 
     if (shutdownOnJvmExit && shutdownHook == null) {
       shutdownHook = new Thread(() -> shutdownPool(true, true));
@@ -325,7 +332,7 @@ private void notifyUp() {
       // check such that we only notify once
       if (!dataSourceUp.get()) {
         dataSourceUp.set(true);
-        startHeartBeatIfStopped();
+        startBackgroundActionsIfStopped();
         dataSourceDownReason = null;
         Log.error("RESOLVED FATAL: DataSource [" + name + "] is back up!");
         if (notify != null) {
@@ -648,6 +655,10 @@ private void shutdownPool(boolean closeBusyConnections, boolean fromHook) {
     stopHeartBeatIfRunning();
     PoolStatus status = queue.shutdown(closeBusyConnections);
     dataSourceUp.set(false);
+
+    // we must stop the executor after queue.shutdown
+    stopAsyncExecutorIfRunning();
+
     if (fromHook) {
       Log.info("DataSource [{0}] shutdown on JVM exit {1}  psc[hit:{2} miss:{3} put:{4} rem:{5}]", name, status, pscHit, pscMiss, pscPut, pscRem);
     } else {
@@ -656,6 +667,8 @@ private void shutdownPool(boolean closeBusyConnections, boolean fromHook) {
     }
   }
 
+
+
   private void removeShutdownHook() {
     if (shutdownHook != null) {
       try {
@@ -684,8 +697,8 @@ public boolean isDataSourceUp() {
     return dataSourceUp.get();
   }
 
-  private void startHeartBeatIfStopped() {
-    heartbeatLock.lock();
+  private void startBackgroundActionsIfStopped() {
+    backgroundtasksLock.lock();
     try {
       // only start if it is not already running
       if (heartBeatTimer == null) {
@@ -695,21 +708,44 @@ private void startHeartBeatIfStopped() {
           heartBeatTimer.scheduleAtFixedRate(new HeartBeatRunnable(), freqMillis, freqMillis);
         }
       }
+      if (executor == null) {
+        this.executor = Executors.newCachedThreadPool();
+      }
     } finally {
-      heartbeatLock.unlock();
+      backgroundtasksLock.unlock();
     }
   }
 
   private void stopHeartBeatIfRunning() {
-    heartbeatLock.lock();
+    backgroundtasksLock.lock();
     try {
       // only stop if it was running
       if (heartBeatTimer != null) {
         heartBeatTimer.cancel();
         heartBeatTimer = null;
       }
     } finally {
-      heartbeatLock.unlock();
+      backgroundtasksLock.unlock();
+    }
+  }
+
+  private void stopAsyncExecutorIfRunning() {
+    backgroundtasksLock.lock();
+    try {
+      // only stop if it was running
+      if (executor != null) {
+        executor.shutdown();
+        try {
+          if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+            Log.warn("DataSource [{0}] could not terminate executor {1}", name, executor);
+          }
+        } catch (InterruptedException ie) {
+          Log.warn("DataSource [{0}] could not terminate executor {1}", name, executor, ie);
+        }
+        executor = null;
+      }
+    } finally {
+      backgroundtasksLock.unlock();
     }
   }
 
@@ -785,6 +821,34 @@ public PoolStatus status(boolean reset) {
     return queue.status(reset);
   }
 
+  /**
+   * Executes the runnable asynchronously. The default grace period is 5 seconds.
+   *
+   * @param runnable
+   * @return true, if the runnable is finished in 5 seconds
+   */
+  boolean execute(Runnable runnable) {
+    backgroundtasksLock.lock();
+    try {
+      if (executor != null) {
+        executor.submit(runnable).get(5, TimeUnit.SECONDS);
+        return true;
+      }
+    } catch (InterruptedException | TimeoutException e) {
+      return false;
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof RuntimeException) {
+        throw (RuntimeException) e.getCause();
+      }
+      throw new RuntimeException(e.getCause());
+    } finally {
+      backgroundtasksLock.unlock();
+    }
+    // execute in sync
+    runnable.run();
+    return true;
+  }
+
   static final class Status implements PoolStatus {
 
     private final int minSize;

ebean-datasource/src/main/java/io/ebean/datasource/pool/PooledConnection.java

@@ -222,6 +222,23 @@ String fullDescription() {
     return "name[" + name + "] startTime[" + startUseTime() + "] busySeconds[" + busySeconds() + "] stackTrace[" + stackTraceAsString() + "] stmt[" + lastStatement() + "]";
   }
 
+  void closeConnectionFully(boolean logErrors) {
+    boolean success = pool.execute(new Runnable() {
+      @Override
+      public void run() {
+        _closeConnectionFully(logErrors);
+      }
+
+      @Override
+      public String toString() {
+        return fullDescription();
+      }
+    });
+    if (!success) {
+      Log.warn("timeout while closing {0}", fullDescription());
+    }
+  }
+
   /**
    * Close the connection fully NOT putting in back into the pool.
    * <p>
@@ -231,7 +248,7 @@ String fullDescription() {
    *
    * @param logErrors if false then don't log errors when closing
    */
-  void closeConnectionFully(boolean logErrors) {
+  void _closeConnectionFully(boolean logErrors) {
     if (Log.isLoggable(System.Logger.Level.TRACE)) {
       Log.trace("Closing Connection[{0}] reason[{1}], pstmtStats: {2}", name, closeReason, pstmtCache.description());
     }

ebean-datasource/src/test/java/io/ebean/datasource/pool/ConnectionPoolHangUpTest.java

@@ -0,0 +1,44 @@
+package io.ebean.datasource.pool;
+
+import io.ebean.datasource.DataSourceBuilder;
+import io.ebean.datasource.DataSourcePool;
+import org.h2.jdbc.JdbcConnection;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+
+public class ConnectionPoolHangUpTest {
+
+  @Test
+  void testHoldLockOnObject() throws Exception {
+    DataSourcePool pool = DataSourceBuilder.create()
+      .url("jdbc:h2:mem:testConnectionPoolHangUp")
+      .username("sa")
+      .password("sa")
+      .heartbeatFreqSecs(1)
+      .minConnections(1)
+      .maxConnections(1)
+      .trimPoolFreqSecs(1)
+      .heartbeatMaxPoolExhaustedCount(0)
+      .failOnStart(false)
+      .build();
+    try {
+      Connection conn = pool.getConnection();
+      Thread t = new Thread(() -> {
+        try {
+          JdbcConnection h2Conn = conn.unwrap(JdbcConnection.class);
+          synchronized (h2Conn) {
+            Thread.sleep(300000);
+          }
+        } catch (Exception e) {
+          // nop
+        }
+      });
+      t.setDaemon(true);
+      t.start();
+    } finally {
+      pool.shutdown();
+    }
+  }
+
+}