Alternate fair/unfair lock modes in the connection pool to improve throughput (#862)

stIncMale · web-flow · commit 146e7ce7e9b4 · 2022-01-27T09:17:32.000-07:00
Another tiny change is moving `pool.release(openConnection)` out of the guarded section in `DefaultConnectionPool.OpenConcurrencyLimiter.tryHandOverOrRelease`: it does not have to be guarded, and when not only fair locking is used, this change has a significant positive measurable effect on the connection cycle (check out/check in) throughput. JAVA-4452
diff --git a/config/spotbugs/exclude.xml b/config/spotbugs/exclude.xml
@@ -232,7 +232,7 @@
     <!-- Non-short-circuit operators are used intentionally. -->
     <Match>
         <Class name="com.mongodb.internal.connection.ConcurrentPool$StateAndPermits"/>
-        <Method name="acquirePermitFair"/>
+        <Method name="acquirePermit"/>
         <Bug pattern="NS_DANGEROUS_NON_SHORT_CIRCUIT"/>
     </Match>
     <Match>
diff --git a/driver-core/src/main/com/mongodb/internal/connection/ConcurrentPool.java b/driver-core/src/main/com/mongodb/internal/connection/ConcurrentPool.java
@@ -28,8 +28,10 @@
 
 import java.util.Iterator;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
@@ -162,7 +164,7 @@ public T get() {
     public T get(final long timeout, final TimeUnit timeUnit) {
         stateAndPermits.throwIfClosedOrPaused();
 
-        if (!stateAndPermits.acquirePermitFair(timeout, timeUnit)) {
+        if (!stateAndPermits.acquirePermit(timeout, timeUnit)) {
             throw new MongoTimeoutException(String.format("Timeout waiting for a pooled item after %d %s", timeout, timeUnit));
         }
 
@@ -221,7 +223,7 @@ public void prune() {
     public void ensureMinSize(final int minSize, final Consumer<T> initAndRelease) {
         stateAndPermits.throwIfClosedOrPaused();
         while (getCount() < minSize) {
-            if (!stateAndPermits.acquirePermitFair(0, TimeUnit.MILLISECONDS)) {
+            if (!stateAndPermits.acquirePermit(0, TimeUnit.MILLISECONDS)) {
                 break;
             }
             initAndRelease.accept(createNewAndReleasePermitIfFailure());
@@ -246,7 +248,7 @@ private T createNewAndReleasePermitIfFailure() {
      */
     @VisibleForTesting(otherwise = PRIVATE)
     boolean acquirePermit(final long timeout, final TimeUnit timeUnit) {
-        return stateAndPermits.acquirePermitFair(timeout, timeUnit);
+        return stateAndPermits.acquirePermit(timeout, timeUnit);
     }
 
     /**
@@ -330,12 +332,42 @@ static boolean isPoolClosedException(final Throwable e) {
     @ThreadSafe
     private static final class StateAndPermits {
         private final Supplier<MongoServerUnavailableException> poolClosedExceptionSupplier;
-        private final ReadWriteLock lock;
+        private final ReentrantReadWriteLock lock;
         private final Condition permitAvailableOrClosedOrPausedCondition;
         private volatile boolean paused;
         private volatile boolean closed;
         private final int maxPermits;
         private volatile int permits;
+        /** When there are not enough available permits to serve all threads requesting a permit, threads are queued and wait on
+         * {@link #permitAvailableOrClosedOrPausedCondition}. Because of this waiting, we want threads to acquire the lock fairly,
+         * to avoid a situation when some threads are sitting in the queue for a long time while others barge in and acquire
+         * the lock without waiting in the queue. Fair locking reduces high percentiles of {@link #acquirePermit(long, TimeUnit)} latencies
+         * but reduces its throughput: it makes latencies roughly equally high for everyone, while keeping them lower than the highest
+         * latencies with unfair locking. The fair approach is in accordance with the
+         * <a href="https://github.com/mongodb/specifications/blob/568093ce7f0e1394cf4952c417e1e7dacc5fef53/source/connection-monitoring-and-pooling/connection-monitoring-and-pooling.rst#waitqueue">
+         * connection pool specification</a>.
+         * <p>
+         * When there are enough available permits to serve all threads requesting a permit, threads still have to acquire the lock,
+         * and still are queued, but since they are not waiting on {@link #permitAvailableOrClosedOrPausedCondition},
+         * threads spend less time in the queue. This results in having smaller high percentiles
+         * of {@link #acquirePermit(long, TimeUnit)} latencies, and we do not want to sacrifice the throughput
+         * to further reduce the high percentiles by acquiring the lock fairly.</p>
+         * <p>
+         * While there is a chance that the expressed reasoning is flawed, it is supported by the results of experiments reported in
+         * comments in <a href="https://jira.mongodb.org/browse/JAVA-4452">JAVA-4452</a>.</p>
+         * <p>
+         * {@link ReentrantReadWriteLock#hasWaiters(Condition)} requires holding the lock to be called, therefore we cannot use it
+         * to discriminate between the two cases described above, and we use {@link #waitersEstimate} instead.
+         * This approach results in sometimes acquiring a lock unfairly when it should have been acquired fairly, and vice versa.
+         * But it appears to be a good enough compromise, that results in having enough throughput when there are enough
+         * available permits and tolerable high percentiles of latencies when there are not enough available permits.</p>
+         * <p>
+         * It may seem viable to use {@link #permits} > 0 as a way to decide that there are likely no waiters,
+         * but benchmarking shows that with this approach high percentiles of contended {@link #acquirePermit(long, TimeUnit)} latencies
+         * (when the number of threads that use the pool is higher than the maximum pool size) become similar to a situation when no
+         * fair locking is used. That is, this approach does not result in the behavior we want.</p>
+         */
+        private final AtomicInteger waitersEstimate;
         @Nullable
         private Supplier<MongoException> causeSupplier;
 
@@ -347,6 +379,7 @@ private static final class StateAndPermits {
             closed = false;
             this.maxPermits = maxPermits;
             permits = maxPermits;
+            waitersEstimate = new AtomicInteger();
             causeSupplier = null;
         }
 
@@ -355,9 +388,7 @@ int permits() {
         }
 
         boolean acquirePermitImmediateUnfair() {
-            if (!lock.writeLock().tryLock()) { // unfair
-                lock.writeLock().lock();
-            }
+            lockUnfair(lock.writeLock());
             try {
                 throwIfClosedOrPaused();
                 if (permits > 0) {
@@ -373,21 +404,24 @@ boolean acquirePermitImmediateUnfair() {
         }
 
         /**
+         * This method also emulates the eager {@link InterruptedException} behavior of
+         * {@link java.util.concurrent.Semaphore#tryAcquire(long, TimeUnit)}.
+         *
          * @param timeout See {@link com.mongodb.internal.Timeout#startNow(long, TimeUnit)}.
          */
-        boolean acquirePermitFair(final long timeout, final TimeUnit unit) throws MongoInterruptedException {
+        boolean acquirePermit(final long timeout, final TimeUnit unit) throws MongoInterruptedException {
             long remainingNanos = unit.toNanos(timeout);
-            try {
-                // preserve the eager InterruptedException behavior of `Semaphore.tryAcquire(long, TimeUnit)`
-                lock.writeLock().lockInterruptibly();
-            } catch (InterruptedException e) {
-                throw new MongoInterruptedException(null, e);
+            if (waitersEstimate.get() == 0) {
+                lockInterruptiblyUnfair(lock.writeLock());
+            } else {
+                lockInterruptibly(lock.writeLock());
             }
             try {
                 while (permits == 0
                         // the absence of short-circuiting is of importance
                         & !throwIfClosedOrPaused()) {
                     try {
+                        waitersEstimate.incrementAndGet();
                         if (timeout < 0 || remainingNanos == Long.MAX_VALUE) {
                             permitAvailableOrClosedOrPausedCondition.await();
                         } else if (remainingNanos >= 0) {
@@ -397,6 +431,8 @@ boolean acquirePermitFair(final long timeout, final TimeUnit unit) throws MongoI
                         }
                     } catch (InterruptedException e) {
                         throw new MongoInterruptedException(null, e);
+                    } finally {
+                        waitersEstimate.decrementAndGet();
                     }
                 }
                 assertTrue(permits > 0);
@@ -409,7 +445,7 @@ boolean acquirePermitFair(final long timeout, final TimeUnit unit) throws MongoI
         }
 
         void releasePermit() {
-            lock.writeLock().lock();
+            lockUnfair(lock.writeLock());
             try {
                 assertTrue(permits < maxPermits);
                 //noinspection NonAtomicOperationOnVolatileField
@@ -421,7 +457,7 @@ void releasePermit() {
         }
 
         void pause(final Supplier<MongoException> causeSupplier) {
-            lock.writeLock().lock();
+            lockUnfair(lock.writeLock());
             try {
                 if (!paused) {
                     this.paused = true;
@@ -435,7 +471,7 @@ void pause(final Supplier<MongoException> causeSupplier) {
 
         void ready() {
             if (paused) {
-                lock.writeLock().lock();
+                lockUnfair(lock.writeLock());
                 try {
                     this.paused = false;
                     this.causeSupplier = null;
@@ -450,7 +486,7 @@ void ready() {
          */
         boolean close() {
             if (!closed) {
-                lock.writeLock().lock();
+                lockUnfair(lock.writeLock());
                 try {
                     if (!closed) {
                         closed = true;
@@ -501,4 +537,45 @@ boolean closed() {
     static String sizeToString(final int size) {
         return size == INFINITE_SIZE ? "infinite" : Integer.toString(size);
     }
+
+    static void lockInterruptibly(final Lock lock) throws MongoInterruptedException {
+        try {
+            lock.lockInterruptibly();
+        } catch (InterruptedException e) {
+            throw new MongoInterruptedException(null, e);
+        }
+    }
+
+    private static void lockInterruptiblyUnfair(final ReentrantReadWriteLock.WriteLock lock) throws MongoInterruptedException {
+        throwIfInterrupted();
+        // `WriteLock.tryLock` is unfair
+        if (!lock.tryLock()) {
+            try {
+                lock.lockInterruptibly();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new MongoInterruptedException(null, new InterruptedException());
+            }
+        }
+    }
+
+    static void lockUnfair(final ReentrantLock lock) {
+        // `ReentrantLock.tryLock` is unfair
+        if (!lock.tryLock()) {
+            lock.lock();
+        }
+    }
+
+    private static void lockUnfair(final ReentrantReadWriteLock.WriteLock lock) {
+        // `WriteLock.tryLock` is unfair
+        if (!lock.tryLock()) {
+            lock.lock();
+        }
+    }
+
+    private static void throwIfInterrupted() throws MongoInterruptedException {
+        if (Thread.currentThread().isInterrupted()) {
+            throw new MongoInterruptedException(null, new InterruptedException());
+        }
+    }
 }
diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java
@@ -93,6 +93,8 @@
 import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
 import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
 import static com.mongodb.internal.connection.ConcurrentPool.INFINITE_SIZE;
+import static com.mongodb.internal.connection.ConcurrentPool.lockInterruptibly;
+import static com.mongodb.internal.connection.ConcurrentPool.lockUnfair;
 import static com.mongodb.internal.connection.ConcurrentPool.sizeToString;
 import static com.mongodb.internal.event.EventListenerHelper.getConnectionPoolListener;
 import static java.lang.String.format;
@@ -831,7 +833,7 @@ public Prune shouldPrune(final UsageTrackingInternalConnection usageTrackingConn
      */
     @ThreadSafe
     private final class OpenConcurrencyLimiter {
-        private final Lock lock;
+        private final ReentrantLock lock;
         private final Condition permitAvailableOrHandedOverOrClosedOrPausedCondition;
         private final int maxPermits;
         private int permits;
@@ -1018,7 +1020,7 @@ private PooledConnection acquirePermitOrGetAvailableOpenedConnection(final boole
         }
 
         private void releasePermit() {
-            lock.lock();
+            lockUnfair(lock);
             try {
                 assertTrue(permits < maxPermits);
                 permits++;
@@ -1059,7 +1061,7 @@ private void giveUpOnTryingToGetAvailableConnection() {
          * from threads that are waiting for a permit to open a connection.
          */
         void tryHandOverOrRelease(final UsageTrackingInternalConnection openConnection) {
-            lock.lock();
+            lockUnfair(lock);
             try {
                 for (//iterate from first (head) to last (tail)
                         MutableReference<PooledConnection> desiredConnectionSlot : desiredConnectionSlots) {
@@ -1073,29 +1075,21 @@ void tryHandOverOrRelease(final UsageTrackingInternalConnection openConnection)
                         return;
                     }
                 }
-                pool.release(openConnection);
             } finally {
                 lock.unlock();
             }
+            pool.release(openConnection);
         }
 
         void signalClosedOrPaused() {
-            lock.lock();
+            lockUnfair(lock);
             try {
                 permitAvailableOrHandedOverOrClosedOrPausedCondition.signalAll();
             } finally {
                 lock.unlock();
             }
         }
 
-        private void lockInterruptibly(final Lock lock) throws MongoInterruptedException {
-            try {
-                lock.lockInterruptibly();
-            } catch (InterruptedException e) {
-                throw new MongoInterruptedException(null, e);
-            }
-        }
-
         /**
          * @param timeoutNanos See {@link Timeout#startNow(long)}.
          * @return The remaining duration as per {@link Timeout#remainingOrInfinite(TimeUnit)} if waiting ended early either
