Remove pool exhausted exception #680

redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java

@@ -27,23 +27,23 @@
 import org.redisson.client.RedisPubSubConnection;
 import org.redisson.config.MasterSlaveServersConfig;
 import org.redisson.misc.RPromise;
+import org.redisson.pubsub.AsyncSemaphore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
-import io.netty.util.concurrent.Promise;
 
 public class ClientConnectionsEntry {
 
     final Logger log = LoggerFactory.getLogger(getClass());
 
     private final Queue<RedisPubSubConnection> allSubscribeConnections = new ConcurrentLinkedQueue<RedisPubSubConnection>();
     private final Queue<RedisPubSubConnection> freeSubscribeConnections = new ConcurrentLinkedQueue<RedisPubSubConnection>();
-    private final AtomicInteger freeSubscribeConnectionsCounter = new AtomicInteger();
+    private final AsyncSemaphore freeSubscribeConnectionsCounter;
 
     private final Queue<RedisConnection> freeConnections = new ConcurrentLinkedQueue<RedisConnection>();
-    private final AtomicInteger freeConnectionsCounter = new AtomicInteger();
+    private final AsyncSemaphore freeConnectionsCounter;
 
     public enum FreezeReason {MANAGER, RECONNECT, SYSTEM}
 
@@ -59,10 +59,10 @@ public enum FreezeReason {MANAGER, RECONNECT, SYSTEM}
     public ClientConnectionsEntry(RedisClient client, int poolMinSize, int poolMaxSize, int subscribePoolMinSize, int subscribePoolMaxSize,
             ConnectionManager connectionManager, NodeType serverMode) {
         this.client = client;
-        this.freeConnectionsCounter.set(poolMaxSize);
+        this.freeConnectionsCounter = new AsyncSemaphore(poolMaxSize);
         this.connectionManager = connectionManager;
         this.nodeType = serverMode;
-        this.freeSubscribeConnectionsCounter.set(subscribePoolMaxSize);
+        this.freeSubscribeConnectionsCounter = new AsyncSemaphore(subscribePoolMaxSize);
 
         if (subscribePoolMaxSize > 0) {
             connectionManager.getConnectionWatcher().add(subscribePoolMinSize, subscribePoolMaxSize, freeSubscribeConnections, freeSubscribeConnectionsCounter);
@@ -107,27 +107,15 @@ public void setFreezed(boolean freezed) {
     }
 
     public int getFreeAmount() {
-        return freeConnectionsCounter.get();
+        return freeConnectionsCounter.getCounter();
     }
 
-    private boolean tryAcquire(AtomicInteger counter) {
-        while (true) {
-            int value = counter.get();
-            if (value == 0) {
-                return false;
-            }
-            if (counter.compareAndSet(value, value - 1)) {
-                return true;
-            }
-        }
-    }
-
-    public boolean tryAcquireConnection() {
-        return tryAcquire(freeConnectionsCounter);
+    public void acquireConnection(Runnable runnable) {
+        freeConnectionsCounter.acquire(runnable);
     }
 
     public void releaseConnection() {
-        freeConnectionsCounter.incrementAndGet();
+        freeConnectionsCounter.release();
     }
 
     public RedisConnection pollConnection() {
@@ -228,12 +216,12 @@ public void releaseSubscribeConnection(RedisPubSubConnection connection) {
         freeSubscribeConnections.add(connection);
     }
 
-    public boolean tryAcquireSubscribeConnection() {
-        return tryAcquire(freeSubscribeConnectionsCounter);
+    public void acquireSubscribeConnection(Runnable runnable) {
+        freeSubscribeConnectionsCounter.acquire(runnable);
     }
 
     public void releaseSubscribeConnection() {
-        freeSubscribeConnectionsCounter.incrementAndGet();
+        freeSubscribeConnectionsCounter.release();
     }
 
     public boolean freezeMaster(FreezeReason reason) {

redisson/src/main/java/org/redisson/connection/IdleConnectionWatcher.java

@@ -19,10 +19,10 @@
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.redisson.client.RedisConnection;
 import org.redisson.config.MasterSlaveServersConfig;
+import org.redisson.pubsub.AsyncSemaphore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,10 +38,10 @@ public static class Entry {
 
         private final int minimumAmount;
         private final int maximumAmount;
-        private final AtomicInteger freeConnectionsCounter;
+        private final AsyncSemaphore freeConnectionsCounter;
         private final Collection<? extends RedisConnection> connections;
 
-        public Entry(int minimumAmount, int maximumAmount, Collection<? extends RedisConnection> connections, AtomicInteger freeConnectionsCounter) {
+        public Entry(int minimumAmount, int maximumAmount, Collection<? extends RedisConnection> connections, AsyncSemaphore freeConnectionsCounter) {
             super();
             this.minimumAmount = minimumAmount;
             this.maximumAmount = maximumAmount;
@@ -84,10 +84,10 @@ public void operationComplete(Future<Void> future) throws Exception {
     }
 
     private boolean validateAmount(Entry entry) {
-        return entry.maximumAmount - entry.freeConnectionsCounter.get() + entry.connections.size() > entry.minimumAmount;
+        return entry.maximumAmount - entry.freeConnectionsCounter.getCounter() + entry.connections.size() > entry.minimumAmount;
     }
 
-    public void add(int minimumAmount, int maximumAmount, Collection<? extends RedisConnection> connections, AtomicInteger freeConnectionsCounter) {
+    public void add(int minimumAmount, int maximumAmount, Collection<? extends RedisConnection> connections, AsyncSemaphore freeConnectionsCounter) {
         entries.add(new Entry(minimumAmount, maximumAmount, connections, freeConnectionsCounter));
     }
 

redisson/src/main/java/org/redisson/connection/pool/ConnectionPool.java

@@ -104,39 +104,51 @@ private void createConnection(final boolean checkFreezed, final AtomicInteger re
             initPromise.tryFailure(cause);
             return;
         }
-
-        RFuture<T> promise = createConnection(entry);
-        promise.addListener(new FutureListener<T>() {
+        
+        acquireConnection(entry, new Runnable() {
+            
             @Override
-            public void operationComplete(Future<T> future) throws Exception {
-                if (future.isSuccess()) {
-                    T conn = future.getNow();
-                    releaseConnection(entry, conn);
-                }
+            public void run() {
+                RFuture<T> promise = createConnection(entry, null);
+                promise.addListener(new FutureListener<T>() {
+                    @Override
+                    public void operationComplete(Future<T> future) throws Exception {
+                        if (future.isSuccess()) {
+                            T conn = future.getNow();
 
-                releaseConnection(entry);
+                            releaseConnection(entry, conn);
+                        }
 
-                if (!future.isSuccess()) {
-                    Throwable cause = new RedisConnectionException(
-                            "Can't init enough connections amount! Only " + (minimumIdleSize - initializedConnections.get()) + " from " + minimumIdleSize + " were initialized. Server: "
-                                                + entry.getClient().getAddr(), future.cause());
-                    initPromise.tryFailure(cause);
-                    return;
-                }
+                        releaseConnection(entry);
 
-                int value = initializedConnections.decrementAndGet();
-                if (value == 0) {
-                    log.info("{} connections initialized for {}", minimumIdleSize, entry.getClient().getAddr());
-                    if (!initPromise.trySuccess(null)) {
-                        throw new IllegalStateException();
-                    }
-                } else if (value > 0 && !initPromise.isDone()) {
-                    if (requests.incrementAndGet() <= minimumIdleSize) {
-                        createConnection(checkFreezed, requests, entry, initPromise, minimumIdleSize, initializedConnections);
+                        if (!future.isSuccess()) {
+                            Throwable cause = new RedisConnectionException(
+                                    "Can't init enough connections amount! Only " + (minimumIdleSize - initializedConnections.get()) + " from " + minimumIdleSize + " were initialized. Server: "
+                                                        + entry.getClient().getAddr(), future.cause());
+                            initPromise.tryFailure(cause);
+                            return;
+                        }
+
+                        int value = initializedConnections.decrementAndGet();
+                        if (value == 0) {
+                            log.info("{} connections initialized for {}", minimumIdleSize, entry.getClient().getAddr());
+                            if (!initPromise.trySuccess(null)) {
+                                throw new IllegalStateException();
+                            }
+                        } else if (value > 0 && !initPromise.isDone()) {
+                            if (requests.incrementAndGet() <= minimumIdleSize) {
+                                createConnection(checkFreezed, requests, entry, initPromise, minimumIdleSize, initializedConnections);
+                            }
+                        }
                     }
-                }
+                });
             }
         });
+
+    }
+
+    protected void acquireConnection(ClientConnectionsEntry entry, Runnable runnable) {
+        entry.acquireConnection(runnable);
     }
 
     protected abstract int getMinimumIdleSize(ClientConnectionsEntry entry);
@@ -148,27 +160,35 @@ protected ClientConnectionsEntry getEntry() {
     public RFuture<T> get() {
         for (int j = entries.size() - 1; j >= 0; j--) {
             ClientConnectionsEntry entry = getEntry();
-            if (!entry.isFreezed() && tryAcquireConnection(entry)) {
-                return connectTo(entry);
+            if (!entry.isFreezed() 
+                    && tryAcquireConnection(entry)) {
+                final RPromise<T> result = connectionManager.newPromise();
+                acquireConnection(entry, new Runnable() {
+                    @Override
+                    public void run() {
+                        connectTo(entry, result);
+                    }
+                });
+                return result;
             }
         }
-
-        List<InetSocketAddress> zeroConnectionsAmount = new LinkedList<InetSocketAddress>();
+        
+        List<InetSocketAddress> failedAttempts = new LinkedList<InetSocketAddress>();
         List<InetSocketAddress> freezed = new LinkedList<InetSocketAddress>();
         for (ClientConnectionsEntry entry : entries) {
             if (entry.isFreezed()) {
                 freezed.add(entry.getClient().getAddr());
             } else {
-                zeroConnectionsAmount.add(entry.getClient().getAddr());
+                failedAttempts.add(entry.getClient().getAddr());
             }
         }
 
-        StringBuilder errorMsg = new StringBuilder(getClass().getSimpleName() + " exhausted! ");
+        StringBuilder errorMsg = new StringBuilder(getClass().getSimpleName() + " no available Redis entries. ");
         if (!freezed.isEmpty()) {
             errorMsg.append(" Disconnected hosts: " + freezed);
         }
-        if (!zeroConnectionsAmount.isEmpty()) {
-            errorMsg.append(" Hosts with fully busy connections: " + zeroConnectionsAmount);
+        if (!failedAttempts.isEmpty()) {
+            errorMsg.append(" Hosts disconnected due to `failedAttempts` limit reached: " + failedAttempts);
         }
 
         RedisConnectionException exception = new RedisConnectionException(errorMsg.toString());
@@ -178,7 +198,9 @@ public RFuture<T> get() {
     public RFuture<T> get(ClientConnectionsEntry entry) {
         if (((entry.getNodeType() == NodeType.MASTER && entry.getFreezeReason() == FreezeReason.SYSTEM) || !entry.isFreezed())
                 && tryAcquireConnection(entry)) {
-            return connectTo(entry);
+            RPromise<T> result = connectionManager.newPromise();
+            connectTo(entry, result);
+            return result;
         }
 
         RedisConnectionException exception = new RedisConnectionException(
@@ -187,7 +209,7 @@ && tryAcquireConnection(entry)) {
     }
 
     protected boolean tryAcquireConnection(ClientConnectionsEntry entry) {
-        return entry.getFailedAttempts() < config.getFailedAttempts() && entry.tryAcquireConnection();
+        return entry.getFailedAttempts() < config.getFailedAttempts();
     }
 
     protected T poll(ClientConnectionsEntry entry) {
@@ -198,21 +220,29 @@ protected RFuture<T> connect(ClientConnectionsEntry entry) {
         return (RFuture<T>) entry.connect();
     }
 
-    private RFuture<T> connectTo(ClientConnectionsEntry entry) {
+    private void connectTo(ClientConnectionsEntry entry, RPromise<T> promise) {
         T conn = poll(entry);
         if (conn != null) {
             if (!conn.isActive()) {
-                return promiseFailure(entry, conn);
+                promiseFailure(entry, promise, conn);
+                return;
             }
 
-            return promiseSuccessful(entry, conn);
+            entry.resetFailedAttempts();
+            promise.trySuccess(conn);
+            return;
         }
 
-        return createConnection(entry);
+        createConnection(entry, promise);
     }
 
-    private RFuture<T> createConnection(final ClientConnectionsEntry entry) {
-        final RPromise<T> promise = connectionManager.newPromise();
+    private RFuture<T> createConnection(final ClientConnectionsEntry entry, RPromise<T> ppromise) {
+        final RPromise<T> promise;
+        if (ppromise != null) {
+            promise = ppromise;
+        } else {
+            promise = connectionManager.newPromise();
+        }
         RFuture<T> connFuture = connect(entry);
         connFuture.addListener(new FutureListener<T>() {
             @Override
@@ -242,11 +272,6 @@ private void connectedSuccessful(ClientConnectionsEntry entry, RPromise<T> promi
         }
     }
 
-    private RFuture<T> promiseSuccessful(ClientConnectionsEntry entry, T conn) {
-        entry.resetFailedAttempts();
-        return (RFuture<T>) conn.getAcquireFuture();
-    }
-
     private void promiseFailure(ClientConnectionsEntry entry, RPromise<T> promise, Throwable cause) {
         if (entry.incFailedAttempts() == config.getFailedAttempts()) {
             checkForReconnect(entry);
@@ -274,23 +299,6 @@ private void promiseFailure(ClientConnectionsEntry entry, RPromise<T> promise, T
         promise.tryFailure(cause);
     }
 
-    private RFuture<T> promiseFailure(ClientConnectionsEntry entry, T conn) {
-        int attempts = entry.incFailedAttempts();
-        if (attempts == config.getFailedAttempts()) {
-            conn.closeAsync();
-            checkForReconnect(entry);
-        } else if (attempts < config.getFailedAttempts()) {
-            releaseConnection(entry, conn);
-        } else {
-            conn.closeAsync();
-        }
-
-        releaseConnection(entry);
-
-        RedisConnectionException cause = new RedisConnectionException(conn + " is not active!");
-        return connectionManager.newFailedFuture(cause);
-    }
-
     private void checkForReconnect(ClientConnectionsEntry entry) {
         if (entry.getNodeType() == NodeType.SLAVE) {
             masterSlaveEntry.slaveDown(entry.getClient().getAddr().getHostName(),

redisson/src/main/java/org/redisson/connection/pool/PubSubConnectionPool.java

@@ -50,10 +50,10 @@ protected RFuture<RedisPubSubConnection> connect(ClientConnectionsEntry entry) {
     }
 
     @Override
-    protected boolean tryAcquireConnection(ClientConnectionsEntry entry) {
-        return entry.tryAcquireSubscribeConnection();
+    protected void acquireConnection(ClientConnectionsEntry entry, Runnable runnable) {
+        entry.acquireSubscribeConnection(runnable);
     }
-
+    
     @Override
     protected void releaseConnection(ClientConnectionsEntry entry) {
         entry.releaseSubscribeConnection();

redisson/src/main/java/org/redisson/pubsub/AsyncSemaphore.java

@@ -74,6 +74,10 @@ public boolean remove(Runnable listener) {
         }
     }
 
+    public int getCounter() {
+        return counter;
+    }
+
     public void release() {
         Runnable runnable = null;