fix: drop per-shard pending borrow counter

Author: dkropachev

driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java

@@ -91,7 +91,6 @@ class HostConnectionPool implements Connection.Owner {
   @VisibleForTesting Set<Connection>[] trash;
 
   private Queue<PendingBorrow>[] pendingBorrows;
-  private AtomicInteger[] pendingBorrowCountByShard;
   final AtomicInteger pendingBorrowCount = new AtomicInteger();
 
   private AtomicInteger[] scheduledForCreation;
@@ -211,12 +210,10 @@ public void tempBlockAdvShardAwareness(long millis) {
   }
 
   int pendingBorrowCountForShard(int shardId) {
-    if (pendingBorrowCountByShard == null
-        || shardId < 0
-        || shardId >= pendingBorrowCountByShard.length) {
+    if (pendingBorrows == null || shardId < 0 || shardId >= pendingBorrows.length) {
       return OperationTimedOutException.UNAVAILABLE;
     }
-    return pendingBorrowCountByShard[shardId].get();
+    return pendingBorrows[shardId].size();
   }
 
   int openConnectionCountForShard(int shardId) {
@@ -331,13 +328,11 @@ ListenableFuture<Void> initAsyncWithConnection(Connection reusedConnection) {
     open = new AtomicInteger[shardsCount];
     trash = new Set[shardsCount];
     pendingBorrows = new Queue[shardsCount];
-    pendingBorrowCountByShard = new AtomicInteger[shardsCount];
     for (int i = 0; i < shardsCount; ++i) {
       this.connections[i] = new CopyOnWriteArrayList<Connection>();
       scheduledForCreation[i] = new AtomicInteger();
       open[i] = new AtomicInteger();
       trash[i] = new CopyOnWriteArraySet<Connection>();
-      pendingBorrowCountByShard[i] = new AtomicInteger();
       pendingBorrows[i] = new ConcurrentLinkedQueue<PendingBorrow>();
     }
 
@@ -653,7 +648,6 @@ private ListenableFuture<Connection> enqueue(
     }
 
     PendingBorrow pendingBorrow = new PendingBorrow(timeout, unit, timeoutsExecutor);
-    pendingBorrowCountByShard[shardId].incrementAndGet();
     pendingBorrows[shardId].add(pendingBorrow);
 
     // If we raced with shutdown, make sure the future will be completed. This has no effect if it
@@ -718,7 +712,6 @@ private void dequeue(final Connection connection) {
         connection.inFlight.decrementAndGet();
       } else {
         pendingBorrowCount.decrementAndGet();
-        pendingBorrowCountByShard[connection.shardId()].decrementAndGet();
         // Ensure that the keyspace set on the connection is the one set on the pool state, in the
         // general case it will be.
         ListenableFuture<Connection> setKeyspaceFuture =

driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTimeoutDiagnosticsTest.java

@@ -18,20 +18,21 @@
 import static org.testng.Assert.assertEquals;
 
 import java.lang.reflect.Field;
-import java.util.AbstractQueue;
-import java.util.Iterator;
+import java.util.ArrayDeque;
 import java.util.Queue;
-import java.util.concurrent.atomic.AtomicInteger;
 import org.objenesis.ObjenesisStd;
 import org.testng.annotations.Test;
 
 public class HostConnectionPoolTimeoutDiagnosticsTest {
 
   @Test(groups = "unit")
-  public void should_use_constant_time_pending_borrow_count_per_shard() throws Exception {
+  public void should_report_pending_borrow_queue_size_per_shard() throws Exception {
     HostConnectionPool pool = allocateInstance(HostConnectionPool.class);
-    setField(pool, "pendingBorrows", new Queue[] {new ThrowingSizeQueue()});
-    setField(pool, "pendingBorrowCountByShard", new AtomicInteger[] {new AtomicInteger(6)});
+    Queue<Object> pendingBorrows = new ArrayDeque<Object>();
+    for (int i = 0; i < 6; i++) {
+      pendingBorrows.add(new Object());
+    }
+    setField(pool, "pendingBorrows", new Queue[] {pendingBorrows});
 
     assertEquals(pool.pendingBorrowCountForShard(0), 6);
   }
@@ -55,32 +56,4 @@ private static void setField(Object target, String name, Object value) throws Ex
     }
     throw new NoSuchFieldException(name);
   }
-
-  private static class ThrowingSizeQueue extends AbstractQueue<Object> {
-
-    @Override
-    public Iterator<Object> iterator() {
-      throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public int size() {
-      throw new AssertionError("pendingBorrowCountForShard must not traverse the queue");
-    }
-
-    @Override
-    public boolean offer(Object o) {
-      throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Object poll() {
-      throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Object peek() {
-      throw new UnsupportedOperationException();
-    }
-  }
 }