Added acquiring to describe how many application threads are waiting for a connection from the pool.

Fixed the bug where the pool status was not showing the correct status of each small pool.

Author: Zhen

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -93,12 +93,19 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
 
         return Futures.asCompletionStage( connectionFuture ).handle( ( channel, error ) ->
         {
-            processAcquisitionError( address, error );
-            assertNotClosed( address, channel, pool );
-            NettyConnection nettyConnection = new NettyConnection( channel, pool, clock, metricsListener );
+            try
+            {
+                processAcquisitionError( address, error );
+                assertNotClosed( address, channel, pool );
+                NettyConnection nettyConnection = new NettyConnection( channel, pool, clock, metricsListener );
 
-            metricsListener.afterAcquiredOrCreated( address, acquireEvent );
-            return nettyConnection;
+                metricsListener.afterAcquiredOrCreated( address, acquireEvent );
+                return nettyConnection;
+            }
+            finally
+            {
+                metricsListener.afterAcquiringOrCreating( address );
+            }
         } );
     }
 
@@ -167,9 +174,9 @@ public CompletionStage<Void> close()
     }
 
     @Override
-    public boolean isOpen()
+    public boolean isOpen( BoltServerAddress address )
     {
-        return !closed.get();
+        return pools.containsKey( address );
     }
 
     private ChannelPool getOrCreatePool( BoltServerAddress address )

driver/src/main/java/org/neo4j/driver/internal/metrics/ConnectionPoolMetricsListener.java

@@ -32,6 +32,8 @@ public interface ConnectionPoolMetricsListener
 
     void beforeAcquiringOrCreating( PoolListenerEvent listenerEvent );
 
+    void afterAcquiringOrCreating();
+
     void afterAcquiredOrCreated( PoolListenerEvent listenerEvent );
 
     void afterTimedOutToAcquireOrCreate();

driver/src/main/java/org/neo4j/driver/internal/metrics/InternalAbstractMetrics.java

@@ -68,6 +68,12 @@ public void beforeAcquiringOrCreating( BoltServerAddress serverAddress, Listener
 
         }
 
+        @Override
+        public void afterAcquiringOrCreating( BoltServerAddress serverAddress )
+        {
+
+        }
+
         @Override
         public void afterAcquiredOrCreated( BoltServerAddress serverAddress, ListenerEvent.PoolListenerEvent acquireEvent )
         {

driver/src/main/java/org/neo4j/driver/internal/metrics/InternalConnectionPoolMetrics.java

@@ -45,6 +45,7 @@ public class InternalConnectionPoolMetrics implements ConnectionPoolMetrics, Con
     private final AtomicLong created = new AtomicLong();
     private final AtomicLong failedToCreate = new AtomicLong();
 
+    private final AtomicInteger acquiring = new AtomicInteger();
     private final AtomicLong acquired = new AtomicLong();
     private final AtomicLong timedOutToAcquire = new AtomicLong();
 
@@ -90,6 +91,13 @@ public void afterClosed()
     public void beforeAcquiringOrCreating( PoolListenerEvent listenerEvent )
     {
         listenerEvent.start();
+        acquiring.incrementAndGet();
+    }
+
+    @Override
+    public void afterAcquiringOrCreating()
+    {
+        acquiring.decrementAndGet();
     }
 
     @Override
@@ -116,7 +124,7 @@ public String uniqueName()
     @Override
     public PoolStatus poolStatus()
     {
-        if ( pool.isOpen() )
+        if ( pool.isOpen( address ) )
         {
             return PoolStatus.Open;
         }
@@ -168,6 +176,12 @@ public long closed()
         return closed.get();
     }
 
+    @Override
+    public int acquiring()
+    {
+        return acquiring.get();
+    }
+
     @Override
     public long acquired()
     {
@@ -183,8 +197,9 @@ public Histogram acquisitionTimeHistogram()
     @Override
     public String toString()
     {
-        return format( "[created=%s, closed=%s, creating=%s, failedToCreate=%s, acquired=%s, " +
+        return format( "[created=%s, closed=%s, creating=%s, failedToCreate=%s, acquiring=%s, acquired=%s, " +
                         "timedOutToAcquire=%s, inUse=%s, idle=%s, poolStatus=%s, acquisitionTimeHistogram=%s]",
-                created(), closed(), creating(), failedToCreate(), acquired(), timedOutToAcquire(), inUse(), idle(), poolStatus(), acquisitionTimeHistogram() );
+                created(), closed(), creating(), failedToCreate(), acquiring(), acquired(),
+                timedOutToAcquire(), inUse(), idle(), poolStatus(), acquisitionTimeHistogram() );
     }
 }

driver/src/main/java/org/neo4j/driver/internal/metrics/InternalMetrics.java

@@ -87,6 +87,12 @@ public void beforeAcquiringOrCreating( BoltServerAddress serverAddress, PoolList
         poolMetrics( serverAddress ).beforeAcquiringOrCreating( listenerEvent );
     }
 
+    @Override
+    public void afterAcquiringOrCreating( BoltServerAddress serverAddress )
+    {
+        poolMetrics( serverAddress ).afterAcquiringOrCreating();
+    }
+
     @Override
     public void afterAcquiredOrCreated( BoltServerAddress serverAddress, PoolListenerEvent listenerEvent )
     {

driver/src/main/java/org/neo4j/driver/internal/metrics/MetricsListener.java

@@ -68,6 +68,12 @@ public interface MetricsListener
      */
     void beforeAcquiringOrCreating( BoltServerAddress serverAddress, PoolListenerEvent acquireEvent );
 
+    /**
+     * After acquiring or creating a new netty channel from pool regardless successfully or not.
+     * @param serverAddress the server the netty channel binds to
+     */
+    void afterAcquiringOrCreating( BoltServerAddress serverAddress );
+
     /**
      * After acquiring or creating a new netty channel from pool successfully.
      * @param serverAddress the server the netty channel binds to

driver/src/main/java/org/neo4j/driver/internal/metrics/spi/ConnectionPoolMetrics.java

@@ -31,30 +31,27 @@ public interface ConnectionPoolMetrics
     String uniqueName();
 
     /**
-     * The status of the pool
+     * The status of the pool.
      * @return The status of the pool.
      */
     PoolStatus poolStatus();
 
     /**
-     * The amount of channels that are in-use (borrowed out of the pool).
-     * The number is changing up and down from time to time.
-     * @return The amount of channels that are in-use
+     * The amount of channels that are currently in-use (borrowed out of the pool).
+     * @return The amount of channels that are currently in-use
      */
     int inUse();
 
     /**
-     * The amount of channels that are idle (buffered inside the pool).
-     * The number is changing up and down from time to time.
-     * @return The amount of channels that are idle.
+     * The amount of channels that are currently idle (buffered inside the pool).
+     * @return The amount of channels that are currently idle.
      */
     int idle();
 
     /**
-     * The amount of channels that are waiting to be created.
-     * The amount is increased by one when the pool noticed a request to create a new connection.
-     * The amount is decreased by one when the pool noticed a new connection is created regardless successfully or not.
-     * The number is changing up and down from time to time.
+     * The amount of channels that are currently waiting to be created.
+     * The amount is increased by one when the pool noticed a request to create a new channel.
+     * The amount is decreased by one when the pool noticed a new channel is created successfully or failed to create.
      * @return The amount of channels that are waiting to be created.
      */
     int creating();
@@ -78,21 +75,30 @@ public interface ConnectionPoolMetrics
     long closed();
 
     /**
-     * An increasing-only number to record how many connections have been acquired from the pool.
+     * The current count of application requests to wait for acquiring a connection from the pool.
+     * The reason to wait could be waiting for creating a new channel, or waiting for a channel to be free by application when the pool is full.
+     * @return The current amount of application request to wait for acquiring a connection from the pool.
+     */
+    int acquiring();
+
+    /**
+     * An increasing-only number to record how many connections have been acquired from the pool
+     * The connections acquired could hold either a newly created channel or a reused channel from the pool.
      * @return The amount of connections that have been acquired from the pool.
      */
     long acquired();
 
     /**
      * An increasing-only number to record how many times that we've failed to acquire a connection from the pool within configured maximum acquisition timeout
      * set by {@link Config.ConfigBuilder#withConnectionAcquisitionTimeout(long, TimeUnit)}.
+     * The connection acquired could hold either a newly created channel or a reused channel from the pool.
      * @return The amount of failures to acquire a connection from the pool within maximum connection acquisition timeout.
      */
     long timedOutToAcquire();
 
     /**
      * An acquisition time histogram records how long it takes to acquire an connection from this pool.
-     * The connection acquired from the pool could either be a connection idling inside the pool or a connection created by the pool.
+     * The connection acquired from the pool could contain either a channel idling inside the pool or a channel created by the pool.
      * @return The acquisition time histogram.
      */
     Histogram acquisitionTimeHistogram();

driver/src/main/java/org/neo4j/driver/internal/spi/ConnectionPool.java

@@ -35,5 +35,5 @@ public interface ConnectionPool
 
     CompletionStage<Void> close();
 
-    boolean isOpen();
+    boolean isOpen( BoltServerAddress address );
 }

driver/src/test/java/org/neo4j/driver/internal/util/FailingConnectionDriverFactory.java

@@ -95,9 +95,9 @@ public CompletionStage<Void> close()
         }
 
         @Override
-        public boolean isOpen()
+        public boolean isOpen( BoltServerAddress address )
         {
-            return delegate.isOpen();
+            return delegate.isOpen( address );
         }
     }