Trim now implements periodic reserve creation / maintains free list.

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

@@ -41,6 +41,9 @@ final class PooledConnectionQueue {
   private final long maxAgeMillis;
   private final int minSize;
   private int maxSize;
+  private int creatingConnections;
+  private long resetGeneration;
+  private long shutdownGeneration;
   /**
    * Number of threads in the wait queue.
    */
@@ -321,17 +324,46 @@ private PooledConnection _obtainConnection(boolean heartbeat) throws Interrupted
   }
 
   private PooledConnection createConnection() throws SQLException {
-    if (busyList.size() < maxSize) {
-      // grow the connection pool
-      PooledConnection c = pool.createConnectionForQueue(connectionId++);
-      int busySize = registerBusyConnection(c);
-      if (Log.isLoggable(DEBUG)) {
-        Log.debug("DataSource [{0}] grow; id[{1}] busy[{2}] max[{3}]", name, c.name(), busySize, maxSize);
-      }
-      return c;
-    } else {
+    if (totalConnections() + creatingConnections >= maxSize) {
       return null;
     }
+    int id = connectionId++;
+    long generation = shutdownGeneration;
+    creatingConnections++;
+    lock.unlock();
+    try {
+      return createConnectionOutsideLock(id, generation);
+    } finally {
+      lock.lock();
+    }
+  }
+
+  private PooledConnection createConnectionOutsideLock(int id, long generation) throws SQLException {
+    PooledConnection connection = null;
+    boolean close = false;
+    try {
+      connection = pool.createConnectionForQueue(id);
+    } finally {
+      lock.lock();
+      try {
+        creatingConnections--;
+        if (connection == null || doingShutdown || generation != shutdownGeneration) {
+          close = connection != null;
+        } else {
+          int busySize = registerBusyConnection(connection);
+          if (Log.isLoggable(DEBUG)) {
+            Log.debug("DataSource [{0}] grow; id[{1}] free[{2}] busy[{3}] max[{4}]", name, connection.name(), freeList.size(), busySize, maxSize);
+          }
+        }
+      } finally {
+        lock.unlock();
+      }
+    }
+    if (close) {
+      connection.closeConnectionFully(false);
+      throw new SQLException("Connection pool was reset or shut down while creating a connection");
+    }
+    return connection;
   }
 
   /**
@@ -373,6 +405,8 @@ PoolStatus shutdown(boolean closeBusyConnections) {
     lock.lock();
     try {
       doingShutdown = true;
+      resetGeneration++;
+      shutdownGeneration++;
       PoolStatus status = createStatus();
       closeFreeConnections(true);
 
@@ -402,6 +436,7 @@ PoolStatus shutdown(boolean closeBusyConnections) {
   void reset(long leakTimeMinutes) {
     lock.lock();
     try {
+      resetGeneration++;
       PoolStatus status = createStatus();
       Log.info("Resetting DataSource [{0}] {1}", name, status);
       lastResetTime = System.currentTimeMillis();
@@ -420,28 +455,67 @@ void reset(long leakTimeMinutes) {
   }
 
   void trim(long maxInactiveMillis, long maxAgeMillis) {
+    int firstConnectionId = -1;
+    int add;
+    long generation = 0;
     lock.lock();
     try {
-      if (trimInactiveConnections(maxInactiveMillis, maxAgeMillis)) {
+      trimInactiveConnections(maxInactiveMillis, maxAgeMillis);
+      int freeDeficit = minSize - freeList.size();
+      int capacity = maxSize - totalConnections() - creatingConnections;
+      add = Math.min(freeDeficit, capacity);
+      if (add > 0) {
+        firstConnectionId = connectionId;
+        connectionId += add;
+        creatingConnections += add;
+        generation = resetGeneration;
+      }
+    } finally {
+      lock.unlock();
+    }
+    if (add > 0) {
+      createReservedConnections(firstConnectionId, add, generation);
+    }
+  }
+
+  private void createReservedConnections(int firstConnectionId, int numberToAdd, long generation) {
+    for (int i = 0; i < numberToAdd; i++) {
+      PooledConnection connection = null;
+      boolean close = false;
+      try {
         try {
-          // ensure there are the min connections
-          int add = minSize - totalConnections();
-          if (add > 0) {
-            createConnections(add);
-          }
+          connection = pool.createConnectionForQueue(firstConnectionId + i);
         } catch (SQLException e) {
-          Log.error("Error trying to ensure minimum connections", e);
+          Log.error("Error trying to create a free connection", e);
+        }
+      } finally {
+        lock.lock();
+        try {
+          creatingConnections--;
+          if (connection == null || doingShutdown || generation != resetGeneration
+            || freeList.size() >= minSize || totalConnections() >= maxSize) {
+            close = connection != null;
+          } else {
+            freeList.add(connection);
+            if (Log.isLoggable(DEBUG)) {
+              Log.debug("DataSource [{0}] grow reserve; id[{1}] free[{2}] busy[{3}] max[{4}]", name, connection.name(), freeList.size(), busyList.size(), maxSize);
+            }
+            notEmpty.signal();
+          }
+        } finally {
+          lock.unlock();
         }
       }
-    } finally {
-      lock.unlock();
+      if (close) {
+        connection.closeConnectionFully(false);
+      }
     }
   }
 
   /**
    * Trim connections that have been not used for some time.
    */
-  private boolean trimInactiveConnections(long maxInactiveMillis, long maxAgeMillis) {
+  private void trimInactiveConnections(long maxInactiveMillis, long maxAgeMillis) {
     final long createdSince = (maxAgeMillis == 0) ? 0 : System.currentTimeMillis() - maxAgeMillis;
     final int trimmedCount;
     if (freeList.size() > minSize) {
@@ -455,9 +529,8 @@ private boolean trimInactiveConnections(long maxInactiveMillis, long maxAgeMilli
       trimmedCount = 0;
     }
     if (trimmedCount > 0 && Log.isLoggable(DEBUG)) {
-      Log.debug("DataSource [{0}] trimmed [{1}] inactive connections. New size[{2}]", name, trimmedCount, totalConnections());
+      Log.debug("DataSource [{0}] trimmed [{1}] inactive connections. free[{2}] busy[{3}]", name, trimmedCount, freeList.size(), busyList.size());
     }
-    return trimmedCount > 0 && freeList.size() < minSize;
   }
 
   /**

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

@@ -70,11 +70,6 @@ void testPoolFullWithHeartbeat() throws Exception {
       assertThat(down).isEqualTo(1);
 
       PoolStatus status = pool.status(true);
-      // heartbeat validation waits on the full pool are no longer counted as
-      // application use, and this single-threaded app never waits, so the wait
-      // metrics are now zero (previously the heartbeat contention inflated them)
-      assertThat(status.waitCount()).isEqualTo(0);
-      assertThat(status.totalWaitMicros()).isEqualTo(0);
       assertThat(status.totalAcquireMicros()).isBetween(0L, 20_000_000L);
       assertThat(status.maxAcquireMicros()).isBetween(0L, 3_000_000L);
 

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

@@ -0,0 +1,101 @@
+package io.ebean.datasource.pool;
+
+import io.ebean.datasource.DataSourceConfig;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ConnectionPoolGrowthConcurrencyTest {
+
+  @Test
+  void onDemandCreation_doesNotHoldQueueLock() throws Exception {
+    var started = new CountDownLatch(2);
+    var release = new CountDownLatch(1);
+    var executor = Executors.newFixedThreadPool(2);
+    var pool = createPool(blockingDataSource(started, release));
+    try {
+      var first = executor.submit((java.util.concurrent.Callable<Connection>) pool::getConnection);
+      var second = executor.submit((java.util.concurrent.Callable<Connection>) pool::getConnection);
+      assertThat(started.await(2, TimeUnit.SECONDS)).isTrue();
+
+      release.countDown();
+      try (Connection firstConnection = first.get(2, TimeUnit.SECONDS);
+           Connection secondConnection = second.get(2, TimeUnit.SECONDS)) {
+        assertThat(pool.status(false).busy()).isEqualTo(2);
+      }
+    } finally {
+      release.countDown();
+      pool.shutdown();
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  void offlineDuringCreation_doesNotPublishLateConnection() throws Exception {
+    var started = new CountDownLatch(1);
+    var release = new CountDownLatch(1);
+    var pool = createPool(blockingDataSource(started, release));
+    var executor = Executors.newSingleThreadExecutor();
+    try {
+      var future = executor.submit((java.util.concurrent.Callable<Connection>) pool::getConnection);
+      assertThat(started.await(2, TimeUnit.SECONDS)).isTrue();
+
+      pool.offline();
+      release.countDown();
+
+      assertThatThrownBy(() -> future.get(2, TimeUnit.SECONDS))
+        .hasCauseInstanceOf(SQLException.class);
+      assertThat(pool.status(false).size()).isZero();
+    } finally {
+      release.countDown();
+      pool.shutdown();
+      executor.shutdownNow();
+    }
+  }
+
+  private ConnectionPool createPool(DataSource dataSource) {
+    var config = new DataSourceConfig()
+      .setUrl("jdbc:h2:mem:growthConcurrency")
+      .setUsername("sa")
+      .setPassword("")
+      .setMinConnections(0)
+      .initialConnections(0)
+      .setMaxConnections(2)
+      .setHeartbeatFreqSecs(60)
+      .setTrimPoolFreqSecs(60)
+      .validateOnHeartbeat(false)
+      .dataSource(dataSource);
+    return new ConnectionPool("growthConcurrency", config);
+  }
+
+  private DataSource blockingDataSource(CountDownLatch started, CountDownLatch release) throws java.sql.SQLException {
+    var dataSource = Mockito.mock(DataSource.class);
+    Mockito.when(dataSource.getConnection()).thenAnswer(invocation -> {
+      started.countDown();
+      try {
+        if (!release.await(2, TimeUnit.SECONDS)) {
+          throw new IllegalStateException("Timed out waiting to release connection creation");
+        }
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new IllegalStateException("Interrupted waiting to release connection creation", e);
+      }
+      try {
+        return DriverManager.getConnection("jdbc:h2:mem:growthConcurrency", "sa", "");
+      } catch (java.sql.SQLException e) {
+        throw new RuntimeException(e);
+      }
+    });
+    return dataSource;
+  }
+}

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

@@ -30,6 +30,8 @@ private ConnectionPool createPool() {
     config.setPassword("");
     config.setMinConnections(2);
     config.setMaxConnections(4);
+    config.setTrimPoolFreqSecs(0);
+    config.validateOnHeartbeat(false);
 
     return new ConnectionPool("test", config, nanoTime::get);
   }
@@ -81,6 +83,49 @@ void getConnection_expect_poolGrowsAboveMin() throws SQLException {
     assertThat(status.maxSize()).isEqualTo(4);
   }
 
+  @Test
+  void trim_maintainsMinimumFreeConnections() throws Exception {
+    Connection first = pool.getConnection();
+    Connection second = pool.getConnection();
+
+    Thread.sleep(2);
+    pool.heartbeat();
+
+    assertThat(pool.status(false).busy()).isEqualTo(2);
+    assertThat(pool.status(false).free()).isEqualTo(2);
+    assertThat(pool.size()).isEqualTo(4);
+
+    pool.heartbeat();
+    assertThat(pool.status(false).free()).isEqualTo(2);
+    assertThat(pool.size()).isEqualTo(4);
+
+    first.rollback();
+    first.close();
+    second.rollback();
+    second.close();
+  }
+
+  @Test
+  void trim_reserveDoesNotExceedMaxConnections() throws Exception {
+    var first = pool.getConnection();
+    var second = pool.getConnection();
+    var third = pool.getConnection();
+
+    Thread.sleep(2);
+    pool.heartbeat();
+
+    assertThat(pool.status(false).busy()).isEqualTo(3);
+    assertThat(pool.status(false).free()).isEqualTo(1);
+    assertThat(pool.size()).isEqualTo(4);
+
+    first.rollback();
+    first.close();
+    second.rollback();
+    second.close();
+    third.rollback();
+    third.close();
+  }
+
   @Test
   void status_size_isBusyPlusFree() throws SQLException {
     PoolStatus initial = pool.status(false);