refactor(pool): add ConnectionStatistics class

tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java

@@ -13,7 +13,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.StampedLock;
 import java.util.function.Consumer;
 
 import io.micrometer.core.instrument.Counter;
@@ -31,6 +31,7 @@
 import io.tarantool.core.protocol.Handlers;
 import io.tarantool.core.protocol.IProtoResponse;
 import io.tarantool.pool.exceptions.PoolClosedException;
+import io.tarantool.pool.exceptions.PoolException;
 
 /**
  * Basic pool implementation.
@@ -102,11 +103,8 @@ public class IProtoClientPoolImpl implements IProtoClientPool {
   /** Options for tuning watchers in connections. */
   private final WatcherOptions watcherOpts;
 
-  /** Count of connections which are invalidated or closed. */
-  private final AtomicInteger unavailable;
-
-  /** Count of connections which are reconnecting */
-  private final AtomicInteger reconnecting;
+  /** Statistics on unavailable and reconnecting and total connections. */
+  private final ConnectionStatistics connectionStatistics;
 
   /** Metrics registry. */
   private final MeterRegistry metricsRegistry;
@@ -129,9 +127,6 @@ public class IProtoClientPoolImpl implements IProtoClientPool {
   /** Time in milliseconds after reconnect will be run. */
   private long reconnectAfter;
 
-  /** Total count of connections initialized in pool. */
-  private int totalSize;
-
   /** Count of successful connection requests. */
   private Counter requestSuccess;
 
@@ -351,10 +346,8 @@ public IProtoClientPoolImpl(
     this.isClosed = new AtomicBoolean(false);
     this.gracefulShutdown = gracefulShutdown;
     this.heartbeatOpts = heartbeatOpts;
-    this.unavailable = new AtomicInteger(0);
-    this.reconnecting = new AtomicInteger(0);
+    this.connectionStatistics = new ConnectionStatistics();
     this.watcherOpts = watcherOpts;
-    this.totalSize = 0;
     this.metricsRegistry = metricsRegistry;
     this.ignoredPacketsHandler = ignoredPacketsHandler;
     this.handlers = handlers;
@@ -391,7 +384,7 @@ public void setGroups(List<InstanceConnectionGroup> clientGroups) {
           shrinkGroup(entries.remove(tag), 0);
         }
       }
-      totalSize = newTotal;
+      this.connectionStatistics.setTotal(newTotal);
     }
   }
 
@@ -501,12 +494,12 @@ public void setConnectTimeout(long timeout) throws IllegalArgumentException {
 
   @Override
   public boolean hasAvailableClients() {
-    return unavailable.get() < totalSize;
+    return this.connectionStatistics.hasAvailableClients();
   }
 
   @Override
   public int availableConnections() {
-    return totalSize - unavailable.get();
+    return this.connectionStatistics.availableConnections();
   }
 
   @Override
@@ -556,8 +549,7 @@ private void expandGroup(List<PoolEntry> connects, InstanceConnectionGroup group
               reconnectAfter,
               heartbeatOpts,
               watcherOpts,
-              unavailable,
-              reconnecting,
+              this.connectionStatistics,
               metricsRegistry,
               ignoredPacketsHandler,
               handlers,
@@ -613,16 +605,19 @@ private void initMetrics() {
     if (this.metricsRegistry == null) {
       return;
     }
-    Gauge.builder("pool.size", () -> totalSize)
+    Gauge.builder("pool.size", this.connectionStatistics::total)
         .description("total number of connections at the moment")
         .register(metricsRegistry);
-    Gauge.builder("pool.available", () -> totalSize - unavailable.get())
+    Gauge.builder("pool.available", this.connectionStatistics::availableConnections)
         .description("total number of connections available at the moment")
         .register(metricsRegistry);
-    Gauge.builder("pool.invalidated", () -> unavailable.get() - reconnecting.get())
-        .description("number of connections unavailable at the moment")
+    Gauge.builder("pool.unavailable", this.connectionStatistics::unavailable)
+        .description("total number of connection unavailable at the moment")
         .register(metricsRegistry);
-    Gauge.builder("pool.reconnecting", reconnecting::get)
+    Gauge.builder("pool.invalidated", this.connectionStatistics::invalidatedConnections)
+        .description("number of connections invalidated at the moment")
+        .register(metricsRegistry);
+    Gauge.builder("pool.reconnecting", this.connectionStatistics::reconnecting)
         .description("number of connections reconnecting at the moment")
         .register(metricsRegistry);
 
@@ -706,4 +701,157 @@ private void incPoolLockedConnectionRequests() {
   public ConnectionFactory getFactory() {
     return factory;
   }
+
+  /** Statistics on unavailable and reconnecting connections. Class is thread-safe. */
+  static final class ConnectionStatistics {
+
+    private final StampedLock lock = new StampedLock();
+
+    /** Count of connections which are invalidated/closed + reconnecting. */
+    private volatile int unavailable = 0;
+
+    /** Count of connections which are reconnecting */
+    private volatile int reconnecting = 0;
+
+    /** Total count of connections initialized in pool. */
+    private volatile int total = 0;
+
+    int unavailable() {
+      return this.unavailable;
+    }
+
+    int reconnecting() {
+      return this.reconnecting;
+    }
+
+    public int availableConnections() {
+      long stamp = lock.tryOptimisticRead();
+
+      int localTotal = this.total;
+      int localUnavailable = this.unavailable;
+
+      if (!lock.validate(stamp)) {
+        stamp = lock.readLock();
+        try {
+          localTotal = this.total;
+          localUnavailable = this.unavailable;
+        } finally {
+          lock.unlockRead(stamp);
+        }
+      }
+
+      final int totalAvailable = localTotal - localUnavailable;
+      if (totalAvailable < 0) {
+        throw new PoolException(
+            String.format(
+                "Invalid pool state! Pool has negative available connections! Total connections:"
+                    + " %s, unavailable: %s, available: %s",
+                localTotal, localUnavailable, totalAvailable)) {};
+      }
+      return totalAvailable;
+    }
+
+    public boolean hasAvailableClients() {
+      long stamp = lock.tryOptimisticRead();
+
+      int localTotal = this.total;
+      int localUnavailable = this.unavailable;
+
+      if (!lock.validate(stamp)) {
+        stamp = lock.readLock();
+        try {
+          localTotal = this.total;
+          localUnavailable = this.unavailable;
+        } finally {
+          lock.unlockRead(stamp);
+        }
+      }
+
+      final int totalAvailable = localTotal - localUnavailable;
+      if (totalAvailable < 0) {
+        throw new PoolException(
+            String.format(
+                "Invalid pool state! Pool has negative available connections! Total connections:"
+                    + " %s, unavailable: %s, available: %s",
+                localTotal, localUnavailable, totalAvailable)) {};
+      }
+      return totalAvailable > 0;
+    }
+
+    public int invalidatedConnections() {
+      long stamp = lock.tryOptimisticRead();
+
+      int localUnavailable = this.unavailable;
+      int localReconnecting = this.reconnecting;
+
+      if (!lock.validate(stamp)) {
+        stamp = lock.readLock();
+        try {
+          localUnavailable = this.unavailable;
+          localReconnecting = this.reconnecting;
+        } finally {
+          lock.unlockRead(stamp);
+        }
+      }
+
+      final int totalInvalidating = localUnavailable - localReconnecting;
+      if (totalInvalidating < 0) {
+        throw new PoolException(
+            String.format(
+                "Invalid pool state! Pool has negative count of invalidated connections!"
+                    + " Unavailable: %s, reconnecting: %s, invalidating: %s",
+                localUnavailable, localReconnecting, totalInvalidating)) {};
+      }
+      return totalInvalidating;
+    }
+
+    void incrementUnavailable() {
+      long stamp = lock.writeLock();
+      try {
+        this.unavailable++;
+      } finally {
+        lock.unlockWrite(stamp);
+      }
+    }
+
+    void decrementUnavailable() {
+      long stamp = lock.writeLock();
+      try {
+        this.unavailable--;
+      } finally {
+        lock.unlockWrite(stamp);
+      }
+    }
+
+    void incrementReconnecting() {
+      long stamp = lock.writeLock();
+      try {
+        this.reconnecting++;
+      } finally {
+        lock.unlockWrite(stamp);
+      }
+    }
+
+    void decrementReconnecting() {
+      long stamp = lock.writeLock();
+      try {
+        this.reconnecting--;
+      } finally {
+        lock.unlockWrite(stamp);
+      }
+    }
+
+    public void setTotal(int newTotal) {
+      long stamp = lock.writeLock();
+      try {
+        this.total = newTotal;
+      } finally {
+        lock.unlockWrite(stamp);
+      }
+    }
+
+    public int total() {
+      return this.total;
+    }
+  }
 }

tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java

@@ -8,7 +8,6 @@
 import java.util.ArrayDeque;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 
@@ -31,6 +30,7 @@
 import io.tarantool.core.protocol.Handlers;
 import io.tarantool.core.protocol.IProtoRequestOpts;
 import io.tarantool.core.protocol.IProtoResponse;
+import io.tarantool.pool.IProtoClientPoolImpl.ConnectionStatistics;
 
 /**
  * Client/connection controller for pool.
@@ -56,19 +56,7 @@ final class PoolEntry {
   /** Deque for sliding window. */
   private final ArrayDeque<Integer> window;
 
-  /**
-   * Atomic variable passed from pool for counting unavailable connections.
-   *
-   * <p>When connection becomes unavailable, PoolEntry increments it.
-   */
-  private final AtomicInteger unavailable;
-
-  /**
-   * Atomic variable passed from pool for counting connections in reconnecting state.
-   *
-   * <p>When connection is being closed and reconnect started, PoolEntry increments it.
-   */
-  private final AtomicInteger reconnecting;
+  private final ConnectionStatistics connectionStatistics;
 
   /**
    * Options for heartbeats.
@@ -213,8 +201,8 @@ final class PoolEntry {
    *     io.tarantool.pool.HeartbeatOpts}
    * @param watcherOpts options for connection watcher, instance of {@link
    *     io.tarantool.core.WatcherOptions}
-   * @param unavailable atomic variable to count unavailable clients
-   * @param reconnecting atomic variable to count client in reconnecting state
+   * @param connectionStatistics statistics on unavailable and reconnecting connections. Class is
+   *     thread-safe.
    * @param registry instance of {@code io.micrometer.core.instrument.MeterRegistry}
    * @param ignoredPacketsHandler callback for accepting packets which were ignored by IProtoClient
    *     connections.
@@ -230,8 +218,7 @@ public PoolEntry(
       long reconnectAfter,
       HeartbeatOpts heartbeatOpts,
       WatcherOptions watcherOpts,
-      AtomicInteger unavailable,
-      AtomicInteger reconnecting,
+      ConnectionStatistics connectionStatistics,
       MeterRegistry registry,
       TripleConsumer<String, Integer, IProtoResponse> ignoredPacketsHandler,
       Handlers handlers,
@@ -277,8 +264,7 @@ public PoolEntry(
     this.reconnectAfter = reconnectAfter;
     this.tag = group.getTag();
     this.timerService = timerService;
-    this.unavailable = unavailable;
-    this.reconnecting = reconnecting;
+    this.connectionStatistics = connectionStatistics;
     this.client.onClose(ConnectionCloseEvent.CLOSE_BY_REMOTE, this::handleConnectError);
     this.client.onClose(ConnectionCloseEvent.CLOSE_BY_SHUTDOWN, this::handleConnectError);
     this.useTupleExtension = useTupleExtension;
@@ -307,7 +293,7 @@ public IProtoClient getClient() {
    */
   public void lock() {
     if (!isLocked) {
-      unavailable.incrementAndGet();
+      this.connectionStatistics.incrementUnavailable();
       isLocked = true;
     }
   }
@@ -320,7 +306,7 @@ public void lock() {
   public void unlock() {
     if (isLocked) {
       stopReconnectTask();
-      unavailable.decrementAndGet();
+      this.connectionStatistics.decrementUnavailable();
       isLocked = false;
     }
   }
@@ -483,7 +469,7 @@ private void handleConnectError(Object r, Throwable exc) {
   private void connectAfter() {
     log.info("reconnect {}/{} after {} ms", tag, index, reconnectAfter);
     if (reconnectTask == null) {
-      reconnecting.incrementAndGet();
+      this.connectionStatistics.incrementReconnecting();
     }
     reconnectTask =
         timerService.newTimeout(
@@ -661,7 +647,7 @@ private void incHeartbeatCounters(int fail) {
   /** Stops reconnecting task if it is active. */
   private void stopReconnectTask() {
     if (reconnectTask != null) {
-      reconnecting.decrementAndGet();
+      this.connectionStatistics.decrementReconnecting();
       reconnectTask.cancel();
       reconnectTask = null;
     }