Enable the background thread to hand over a newly opened connection to a thread that is blocked by the max connecting limit (#805)

Backport JAVA-4316, JAVA-4346

Author: stIncMale

driver-core/src/main/com/mongodb/internal/connection/ConcurrentPool.java

@@ -194,21 +194,18 @@ public void prune() {
 
     /**
      * Try to populate this pool with items so that {@link #getCount()} is not smaller than {@code minSize}.
-     * The {@code postCreate} action throwing a exception causes this method to stop and re-throw that exception.
+     * The {@code initAndRelease} action throwing an exception causes this method to stop and re-throw that exception.
      *
-     * @param initialize An action applied to non-{@code null} new items.
-     *                   If an exception is thrown by the action, the action must treat the provided item as if obtained via
-     *                   a {@link #get(long, TimeUnit) get…} method, {@linkplain #release(Object, boolean) releasing} it
-     *                   if an exception is thrown; otherwise the action must not release the item.
+     * @param initAndRelease An action applied to non-{@code null} new items.
+     * If an exception is thrown by the action, the action must {@linkplain #release(Object, boolean) prune} the item.
+     * Otherwise, the action must {@linkplain #release(Object) release} the item.
      */
-    public void ensureMinSize(final int minSize, final Consumer<T> initialize) {
+    public void ensureMinSize(final int minSize, final Consumer<T> initAndRelease) {
         while (getCount() < minSize) {
             if (!acquirePermit(0, TimeUnit.MILLISECONDS)) {
                 break;
             }
-            T newItem = createNewAndReleasePermitIfFailure();
-            initialize.accept(newItem);
-            release(newItem);
+            initAndRelease.accept(createNewAndReleasePermitIfFailure());
         }
     }
 

driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java

@@ -73,6 +73,7 @@
 
 import static com.mongodb.assertions.Assertions.assertFalse;
 import static com.mongodb.assertions.Assertions.assertNotNull;
+import static com.mongodb.assertions.Assertions.assertNull;
 import static com.mongodb.assertions.Assertions.assertTrue;
 import static com.mongodb.assertions.Assertions.fail;
 import static com.mongodb.assertions.Assertions.isTrue;
@@ -198,7 +199,7 @@ public void getAsync(final SingleResultCallback<InternalConnection> callback) {
                         LOGGER.trace(format("Pooled connection %s to server %s is not yet open",
                                 getId(connection), serverId));
                     }
-                    openConcurrencyLimiter.openAsyncOrGetAvailable(connection, timeout, eventSendingCallback);
+                    openConcurrencyLimiter.openAsyncWithConcurrencyLimit(connection, timeout, eventSendingCallback);
                 }
             }
         }));
@@ -358,7 +359,7 @@ public synchronized void run() {
                                 LOGGER.debug(format("Ensuring minimum pooled connections to %s", serverId.getAddress()));
                             }
                             pool.ensureMinSize(settings.getMinSize(), newConnection ->
-                                    openConcurrencyLimiter.openImmediately(new PooledConnection(newConnection)));
+                                    openConcurrencyLimiter.openImmediatelyAndTryHandOverOrRelease(new PooledConnection(newConnection)));
                         }
                     } catch (MongoInterruptedException | MongoTimeoutException e) {
                         //complete the maintenance task
@@ -802,47 +803,56 @@ private final class OpenConcurrencyLimiter {
         }
 
         PooledConnection openOrGetAvailable(final PooledConnection connection, final Timeout timeout) throws MongoTimeoutException {
-            return openOrGetAvailable(connection, true, timeout);
+            PooledConnection result = openWithConcurrencyLimit(connection, OpenWithConcurrencyLimitMode.TRY_GET_AVAILABLE, timeout);
+            return assertNotNull(result);
         }
 
-        void openImmediately(final PooledConnection connection) throws MongoTimeoutException {
-            PooledConnection result = openOrGetAvailable(connection, false, Timeout.immediate());
-            assertTrue(result == connection);
+        void openImmediatelyAndTryHandOverOrRelease(final PooledConnection connection) throws MongoTimeoutException {
+            assertNull(openWithConcurrencyLimit(connection, OpenWithConcurrencyLimitMode.TRY_HAND_OVER_OR_RELEASE, Timeout.immediate()));
         }
 
         /**
          * This method can be thought of as operating in two phases.
          * In the first phase it tries to synchronously acquire a permit to open the {@code connection}
-         * or get a different {@linkplain PooledConnection#opened() opened} connection if {@code tryGetAvailable} is {@code true} and
-         * one becomes available while waiting for a permit.
+         * or get a different {@linkplain PooledConnection#opened() opened} connection if {@code mode} is
+         * {@link OpenWithConcurrencyLimitMode#TRY_GET_AVAILABLE} and one becomes available while waiting for a permit.
          * The first phase has one of the following outcomes:
          * <ol>
          *     <li>A {@link MongoTimeoutException} or a different {@link Exception} is thrown,
          *     and the specified {@code connection} is {@linkplain PooledConnection#closeSilently() silently closed}.</li>
          *     <li>An opened connection different from the specified one is returned,
-         *     and the specified {@code connection} is {@linkplain PooledConnection#closeSilently() silently closed}.</li>
+         *     and the specified {@code connection} is {@linkplain PooledConnection#closeSilently() silently closed}.
+         *     This outcome is possible only if {@code mode} is {@link OpenWithConcurrencyLimitMode#TRY_GET_AVAILABLE}.</li>
          *     <li>A permit is acquired, {@link #connectionCreated(ConnectionPoolListener, ConnectionId)} is reported
          *     and an attempt to open the specified {@code connection} is made. This is the second phase in which
          *     the {@code connection} is {@linkplain PooledConnection#open() opened synchronously}.
          *     The attempt to open the {@code connection} has one of the following outcomes
-         *     combined with releasing the acquired permit:</li>
+         *     combined with releasing the acquired permit:
          *     <ol>
          *         <li>An {@link Exception} is thrown
          *         and the {@code connection} is {@linkplain PooledConnection#closeAndHandleOpenFailure() closed}.</li>
-         *         <li>The specified {@code connection}, which is now opened, is returned.</li>
+         *         <li>Else if the specified {@code connection} is opened successfully and
+         *         {@code mode} is {@link OpenWithConcurrencyLimitMode#TRY_HAND_OVER_OR_RELEASE},
+         *         then {@link #tryHandOverOrRelease(UsageTrackingInternalConnection)} is called and {@code null} is returned.</li>
+         *         <li>Else the specified {@code connection}, which is now opened, is returned.</li>
          *     </ol>
+         *     </li>
          * </ol>
          *
          * @param timeout Applies only to the first phase.
          * @return An {@linkplain PooledConnection#opened() opened} connection which is
-         * either the specified {@code connection} or a different one.
+         * either the specified {@code connection},
+         * or potentially a different one if {@code mode} is {@link OpenWithConcurrencyLimitMode#TRY_GET_AVAILABLE},
+         * or {@code null} if {@code mode} is {@link OpenWithConcurrencyLimitMode#TRY_HAND_OVER_OR_RELEASE}.
          * @throws MongoTimeoutException If the first phase timed out.
          */
-        private PooledConnection openOrGetAvailable(
-                final PooledConnection connection, final boolean tryGetAvailable, final Timeout timeout) throws MongoTimeoutException {
+        @Nullable
+        private PooledConnection openWithConcurrencyLimit(final PooledConnection connection, final OpenWithConcurrencyLimitMode mode,
+                final Timeout timeout) throws MongoTimeoutException {
             PooledConnection availableConnection;
             try {//phase one
-                availableConnection = acquirePermitOrGetAvailableOpenedConnection(tryGetAvailable, timeout);
+                availableConnection = acquirePermitOrGetAvailableOpenedConnection(
+                        mode == OpenWithConcurrencyLimitMode.TRY_GET_AVAILABLE, timeout);
             } catch (RuntimeException e) {
                 connection.closeSilently();
                 throw e;
@@ -853,25 +863,32 @@ private PooledConnection openOrGetAvailable(
             } else {//acquired a permit, phase two
                 try {
                     connection.open();
+                    if (mode == OpenWithConcurrencyLimitMode.TRY_HAND_OVER_OR_RELEASE) {
+                        tryHandOverOrRelease(connection.wrapped);
+                        return null;
+                    } else {
+                        return connection;
+                    }
                 } finally {
                     releasePermit();
                 }
-                return connection;
             }
         }
 
         /**
-         * This method is similar to {@link #openOrGetAvailable(PooledConnection, boolean, Timeout)} with the following differences:
+         * This method is similar to {@link #openWithConcurrencyLimit(PooledConnection, OpenWithConcurrencyLimitMode, Timeout)}
+         * with the following differences:
          * <ul>
-         *     <li>It does not have the {@code tryGetAvailable} parameter and acts as if this parameter were {@code true}.</li>
+         *     <li>It does not have the {@code mode} parameter and acts as if this parameter were
+         *     {@link OpenWithConcurrencyLimitMode#TRY_GET_AVAILABLE}.</li>
          *     <li>While the first phase is still synchronous, the {@code connection} is
          *     {@linkplain PooledConnection#openAsync(SingleResultCallback) opened asynchronously} in the second phase.</li>
          *     <li>Instead of returning a result or throwing an exception via Java {@code return}/{@code throw} statements,
          *     it calls {@code callback.}{@link SingleResultCallback#onResult(Object, Throwable) onResult(result, failure)}
          *     and passes either a {@link PooledConnection} or an {@link Exception}.</li>
          * </ul>
          */
-        void openAsyncOrGetAvailable(
+        void openAsyncWithConcurrencyLimit(
                 final PooledConnection connection, final Timeout timeout, final SingleResultCallback<InternalConnection> callback) {
             PooledConnection availableConnection;
             try {//phase one
@@ -1044,6 +1061,14 @@ private long awaitNanos(final Condition condition, final long timeoutNanos) thro
         }
     }
 
+    /**
+     * @see OpenConcurrencyLimiter#openWithConcurrencyLimit(PooledConnection, OpenWithConcurrencyLimitMode, Timeout)
+     */
+    private enum OpenWithConcurrencyLimitMode {
+        TRY_GET_AVAILABLE,
+        TRY_HAND_OVER_OR_RELEASE
+    }
+
     @NotThreadSafe
     private static final class MutableReference<T> {
         @Nullable

driver-core/src/test/resources/connection-monitoring-and-pooling/pool-checkout-minPoolSize-connection-maxConnecting.json

@@ -0,0 +1,84 @@
+{
+  "version": 1,
+  "style": "integration",
+  "description": "threads blocked by maxConnecting check out minPoolSize connections",
+  "runOn": [
+    {
+      "minServerVersion": "4.4.0"
+    }
+  ],
+  "failPoint": {
+    "configureFailPoint": "failCommand",
+    "mode": "alwaysOn",
+    "data": {
+      "failCommands": [
+        "isMaster",
+        "hello"
+      ],
+      "closeConnection": false,
+      "blockConnection": true,
+      "blockTimeMS": 500
+    }
+  },
+  "poolOptions": {
+    "minPoolSize": 2,
+    "maxPoolSize": 3,
+    "waitQueueTimeoutMS": 5000
+  },
+  "operations": [
+    {
+      "name": "start",
+      "target": "thread1"
+    },
+    {
+      "name": "start",
+      "target": "thread2"
+    },
+    {
+      "name": "wait",
+      "ms": 200
+    },
+    {
+      "name": "checkOut",
+      "thread": "thread1"
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionCreated",
+      "count": 2
+    },
+    {
+      "name": "checkOut",
+      "thread": "thread2"
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionCheckedOut",
+      "count": 2
+    }
+  ],
+  "events": [
+    {
+      "type": "ConnectionCreated",
+      "address": 42
+    },
+    {
+      "type": "ConnectionCreated",
+      "address": 42
+    },
+    {
+      "type": "ConnectionCheckedOut",
+      "address": 42
+    },
+    {
+      "type": "ConnectionCheckedOut",
+      "address": 42
+    }
+  ],
+  "ignore": [
+    "ConnectionClosed",
+    "ConnectionReady",
+    "ConnectionPoolCreated",
+    "ConnectionCheckOutStarted"
+  ]
+}

driver-core/src/test/unit/com/mongodb/internal/connection/ConcurrentPoolTest.java

@@ -141,37 +141,40 @@ public void testCloseItemOnReleaseAfterPoolClosed() {
     @Test
     public void testEnsureMinSize() {
         pool = new ConcurrentPool<TestCloseable>(3, new TestItemFactory());
-        Consumer<TestCloseable> noInit = x -> {};
-        pool.ensureMinSize(0, noInit);
+        Consumer<TestCloseable> initAndRelease = connection -> pool.release(connection);
+        pool.ensureMinSize(0, initAndRelease);
         assertEquals(0, pool.getAvailableCount());
 
-        pool.ensureMinSize(1, noInit);
+        pool.ensureMinSize(1, initAndRelease);
         assertEquals(1, pool.getAvailableCount());
 
-        pool.ensureMinSize(1, noInit);
+        pool.ensureMinSize(1, initAndRelease);
         assertEquals(1, pool.getAvailableCount());
 
         pool.get();
-        pool.ensureMinSize(1, noInit);
+        pool.ensureMinSize(1, initAndRelease);
         assertEquals(0, pool.getAvailableCount());
 
-        pool.ensureMinSize(4, noInit);
+        pool.ensureMinSize(4, initAndRelease);
         assertEquals(3, pool.getAvailableCount());
     }
 
     @Test
     public void whenEnsuringMinSizeShouldNotInitializePooledItemIfNotRequested() {
         pool = new ConcurrentPool<TestCloseable>(3, new TestItemFactory());
 
-        pool.ensureMinSize(1, noInit -> {});
+        pool.ensureMinSize(1, pool::release);
         assertFalse(pool.get().isInitialized());
     }
 
     @Test
     public void whenEnsuringMinSizeShouldInitializePooledItemIfRequested() {
         pool = new ConcurrentPool<TestCloseable>(3, new TestItemFactory());
 
-        pool.ensureMinSize(1, TestCloseable.INIT_ACTION);
+        pool.ensureMinSize(1, connection -> {
+            connection.initialized = true;
+            pool.release(connection);
+        });
         assertTrue(pool.get().isInitialized());
     }
 
@@ -180,7 +183,7 @@ public void testThatEnsuringMinSizeReleasesPermitIfCreateFails() {
         pool = new ConcurrentPool<TestCloseable>(1, new TestItemFactory(true));
 
         try {
-            pool.ensureMinSize(1, TestCloseable.INIT_ACTION);
+            pool.ensureMinSize(1, ignore -> fail());
             fail();
         } catch (MongoException e) {
             // expected
@@ -252,10 +255,6 @@ public ConcurrentPool.Prune shouldPrune(final TestCloseable testCloseable) {
     }
 
     static class TestCloseable implements Closeable {
-        private static final Consumer<TestCloseable> INIT_ACTION = connection -> {
-            connection.initialized = true;
-        };
-
         private boolean closed;
         private ConcurrentPool.Prune shouldPrune;
         private boolean initialized;