Refactor DefaultConnectionPool to use single worker thread when checking out asynchronously (#722)

When implementing 73e3085
I correctly explained the deadlock situation:
1) An async task that created a non-open connection and tries to open it (let us call it an "open task")
may block another async task that tries to create a connection (let us call it a "get task")
if `maxSize` is reached.
2) If both tasks are submitted to the same queue such that the get task is ahead of the open task, then
such a get task naturally blocks the open task while being blocked by the open task.
but failed to optimally fix it.

The solution to the problem I used is to submit tasks to different queues, which,
because both tasks are blocking, required introducing one more async worker thread.

However, there is another approach to solving the deadlock problem,
which did not occur to me at that time: prevent the get task to be ahead of the open task
in the queue. This may happen due to the optimization in the `getAsync` method, which
tries to check out synchronously if an open connection is available,
i.e., if checking it out requires no blocking. It is a nice optimization that I
tried to preserve (it was there before my changes) when implemented JAVA-3927.
However, this optimization opens up opportunities for a get task being blocked
by an open task to be ordered ahead of the open task in the queue.

If I realized that previously, I would have sacrificed this optimization
to eliminate the deadlock instead of introducing another queue and a thread.
Fortunately, the it occurred to me now.

This change also fixes reactive `MainTransactionsTest` runner
so that the `abort.json`: "implicit abort" test succeeds.
This test expects command_started_event for the abortTransaction command, which is sent
as a result of closing the session used by the test. The close method is asynchronous and
the test must wait for its completion.

JAVA-4146

Author: stIncMale

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

@@ -141,7 +141,7 @@ public T get() {
     @Override
     public T get(final long timeout, final TimeUnit timeUnit) {
         if (closed) {
-            throw new IllegalStateException("The pool is closed");
+            throw poolClosedException();
         }
 
         if (!acquirePermit(timeout, timeUnit)) {
@@ -294,4 +294,8 @@ private void close(final T t) {
             // ItemFactory.close() really should not throw
         }
     }
+
+    static IllegalStateException poolClosedException() {
+        return new IllegalStateException("The pool is closed");
+    }
 }

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

@@ -53,10 +53,12 @@
 import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -66,6 +68,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.StampedLock;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 import static com.mongodb.assertions.Assertions.assertFalse;
 import static com.mongodb.assertions.Assertions.assertNotNull;
@@ -91,7 +95,7 @@ class DefaultConnectionPool implements ConnectionPool {
     private final ConnectionPoolSettings settings;
     private final AtomicInteger generation = new AtomicInteger(0);
     private final ScheduledExecutorService sizeMaintenanceTimer;
-    private final Workers workers;
+    private final AsyncWorkManager asyncWorkManager;
     private final Runnable maintenanceTask;
     private final ConnectionPoolListener connectionPoolListener;
     private final ServerId serverId;
@@ -113,7 +117,7 @@ class DefaultConnectionPool implements ConnectionPool {
         sizeMaintenanceTimer = createMaintenanceTimer();
         connectionPoolCreated(connectionPoolListener, serverId, settings);
         openConcurrencyLimiter = new OpenConcurrencyLimiter(MAX_CONNECTING);
-        workers = new Workers();
+        asyncWorkManager = new AsyncWorkManager();
         connectionGenerationSupplier = new ConnectionGenerationSupplier() {
             @Override
             public int getGeneration() {
@@ -172,35 +176,32 @@ public void getAsync(final SingleResultCallback<InternalConnection> callback) {
                 errHandlingCallback.onResult(null, checkOutFailed(failure));
             }
         };
-        PooledConnection immediateConnection = null;
-
-        try {
-            immediateConnection = getPooledConnection(Timeout.immediate());
-        } catch (MongoTimeoutException e) {
-            // fall through
-        } catch (RuntimeException e) {
-            eventSendingCallback.onResult(null, e);
-            return;
-        }
-
-        if (immediateConnection != null) {
-            openAsync(immediateConnection, timeout, eventSendingCallback);
-        } else {
-            workers.getter().execute(() -> {
-                if (timeout.expired()) {
-                    eventSendingCallback.onResult(null, createTimeoutException(timeout));
-                    return;
-                }
+        asyncWorkManager.enqueue(new Task(timeout, t -> {
+            if (t != null) {
+                eventSendingCallback.onResult(null, t);
+            } else {
                 PooledConnection connection;
                 try {
                     connection = getPooledConnection(timeout);
                 } catch (RuntimeException e) {
                     eventSendingCallback.onResult(null, e);
                     return;
                 }
-                openAsync(connection, timeout, eventSendingCallback);
-            });
-        }
+                if (connection.opened()) {
+                    if (LOGGER.isTraceEnabled()) {
+                        LOGGER.trace(format("Pooled connection %s to server %s is already open",
+                                getId(connection), serverId));
+                    }
+                    eventSendingCallback.onResult(connection, null);
+                } else {
+                    if (LOGGER.isTraceEnabled()) {
+                        LOGGER.trace(format("Pooled connection %s to server %s is not yet open",
+                                getId(connection), serverId));
+                    }
+                    openConcurrencyLimiter.openAsyncOrGetAvailable(connection, timeout, eventSendingCallback);
+                }
+            }
+        }));
     }
 
     /**
@@ -223,23 +224,6 @@ private Throwable checkOutFailed(final Throwable t) {
         return result;
     }
 
-    private void openAsync(final PooledConnection pooledConnection, final Timeout timeout,
-                           final SingleResultCallback<InternalConnection> callback) {
-        if (pooledConnection.opened()) {
-            if (LOGGER.isTraceEnabled()) {
-                LOGGER.trace(format("Pooled connection %s to server %s is already open",
-                        getId(pooledConnection), serverId));
-            }
-            callback.onResult(pooledConnection, null);
-        } else {
-            if (LOGGER.isTraceEnabled()) {
-                LOGGER.trace(format("Pooled connection %s to server %s is not yet open",
-                        getId(pooledConnection), serverId));
-            }
-            workers.opener().execute(() -> openConcurrencyLimiter.openAsyncOrGetAvailable(pooledConnection, timeout, callback));
-        }
-    }
-
     @Override
     public void invalidate() {
         LOGGER.debug("Invalidating the connection pool");
@@ -264,7 +248,7 @@ public void close() {
             if (sizeMaintenanceTimer != null) {
                 sizeMaintenanceTimer.shutdownNow();
             }
-            workers.close();
+            asyncWorkManager.close();
             closed = true;
             connectionPoolListener.connectionPoolClosed(new ConnectionPoolClosedEvent(serverId));
         }
@@ -1170,55 +1154,160 @@ int getNumPinnedToTransaction() {
         }
     }
 
+    /**
+     * This class maintains threads needed to perform {@link #getAsync(SingleResultCallback)}.
+     */
     @ThreadSafe
-    private static class Workers implements AutoCloseable {
-        private volatile ExecutorService getter;
-        private volatile ExecutorService opener;
+    private static class AsyncWorkManager implements AutoCloseable {
+        private volatile State state;
+        private volatile BlockingQueue<Task> tasks;
         private final Lock lock;
+        @Nullable
+        private ExecutorService worker;
 
-        Workers() {
+        AsyncWorkManager() {
+            state = State.NEW;
+            tasks = new LinkedBlockingQueue<>();
             lock = new StampedLock().asWriteLock();
         }
 
-        Executor getter() {
-            if (getter == null) {
-                lock.lock();
-                try {
-                    if (getter == null) {
-                        getter = Executors.newSingleThreadExecutor(new DaemonThreadFactory("AsyncGetter"));
-                    }
-                } finally {
-                    lock.unlock();
+        void enqueue(final Task task) {
+            lock.lock();
+            try {
+                if (initUnlessClosed()) {
+                    tasks.add(task);
+                    return;
                 }
+            } finally {
+                lock.unlock();
             }
-            return getter;
+            task.failAsClosed();
         }
 
-        Executor opener() {
-            if (opener == null) {
-                lock.lock();
-                try {
-                    if (opener == null) {
-                        opener = Executors.newSingleThreadExecutor(new DaemonThreadFactory("AsyncOpener"));
-                    }
-                } finally {
-                    lock.unlock();
-                }
+        /**
+         * Invocations of this method must be guarded by {@link #lock}.
+         *
+         * @return {@code false} iff the {@link #state} is {@link State#CLOSED}.
+         */
+        private boolean initUnlessClosed() {
+            boolean result = true;
+            if (state == State.NEW) {
+                worker = Executors.newSingleThreadExecutor(new DaemonThreadFactory("AsyncGetter"));
+                worker.submit(() -> runAndLogUncaught(this::workerRun));
+                state = State.INITIALIZED;
+            } else if (state == State.CLOSED) {
+                result = false;
             }
-            return opener;
+            return result;
         }
 
+        /**
+         * {@linkplain Thread#interrupt() Interrupts} all workers and causes queued tasks to
+         * {@linkplain Task#failAsClosed() fail} asynchronously.
+         */
         @Override
+        @SuppressWarnings("try")
         public void close() {
+            lock.lock();
             try {
-                if (getter != null) {
-                    getter.shutdownNow();
+                if (state != State.CLOSED) {
+                    state = State.CLOSED;
+                    if (worker != null) {
+                        worker.shutdownNow(); // at this point we interrupt `worker`s thread
+                    }
                 }
             } finally {
-                if (opener != null) {
-                    opener.shutdownNow();
+                lock.unlock();
+            }
+        }
+
+        private void workerRun() {
+            try {
+                while (state != State.CLOSED) {
+                    try {
+                        Task task = tasks.take();
+                        if (task.timeout().expired()) {
+                            task.failAsTimedOut();
+                        } else {
+                            task.execute();
+                        }
+                    } catch (RuntimeException e) {
+                        LOGGER.error(null, e);
+                    }
                 }
+            } catch (InterruptedException closed) {
+                // fail the rest of the tasks and stop
             }
+            failAllTasksAfterClosing();
+        }
+
+        private void failAllTasksAfterClosing() {
+            Queue<Task> localGets;
+            lock.lock();
+            try {
+                assertTrue(state == State.CLOSED);
+                // at this point it is guaranteed that no thread enqueues a task
+                localGets = tasks;
+                if (!tasks.isEmpty()) {
+                    tasks = new LinkedBlockingQueue<>();
+                }
+            } finally {
+                lock.unlock();
+            }
+            localGets.forEach(Task::failAsClosed);
+            localGets.clear();
+        }
+
+        private void runAndLogUncaught(final Runnable runnable) {
+            try {
+                runnable.run();
+            } catch (Throwable t) {
+                LOGGER.error("The pool is not going to work correctly from now on. You may want to recreate the MongoClient", t);
+                throw t;
+            }
+        }
+
+        private enum State {
+            NEW,
+            INITIALIZED,
+            CLOSED
+        }
+    }
+
+    /**
+     * An action that is allowed to be completed (failed or executed) at most once, and a timeout associated with it.
+     */
+    @NotThreadSafe
+    final class Task {
+        private final Timeout timeout;
+        private final Consumer<RuntimeException> action;
+        private boolean completed;
+
+        Task(final Timeout timeout, final Consumer<RuntimeException> action) {
+            this.timeout = timeout;
+            this.action = action;
+        }
+
+        void execute() {
+            doComplete(() -> null);
+        }
+
+        void failAsClosed() {
+            doComplete(ConcurrentPool::poolClosedException);
+        }
+
+        void failAsTimedOut() {
+            doComplete(() -> createTimeoutException(timeout));
+        }
+
+        private void doComplete(final Supplier<RuntimeException> failureSupplier) {
+            assertFalse(completed);
+            completed = true;
+            action.accept(failureSupplier.get());
+        }
+
+        Timeout timeout() {
+            return timeout;
         }
     }
 }