diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java index 953c1bb2..4ea6dd22 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java +++ b/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 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 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; + } + } } diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index a1e0067f..dfc4d175 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/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 window; - /** - * Atomic variable passed from pool for counting unavailable connections. - * - *

When connection becomes unavailable, PoolEntry increments it. - */ - private final AtomicInteger unavailable; - - /** - * Atomic variable passed from pool for counting connections in reconnecting state. - * - *

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 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; }