Fixed - AsyncSemaphore doesn't skip canceled tasks in the same thread. #4215

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

@@ -148,8 +148,8 @@ public class ClientConnectionsEntry {
         freeSubscribeConnectionsCounter.removeListeners();
     }
 
-    public void acquireConnection(Runnable runnable, RedisCommand<?> command) {
-        freeConnectionsCounter.acquire(runnable);
+    public CompletableFuture<Void> acquireConnection(RedisCommand<?> command) {
+        return freeConnectionsCounter.acquire();
     }
     
     public void releaseConnection() {
@@ -263,8 +263,8 @@ public class ClientConnectionsEntry {
         freeSubscribeConnections.add(connection);
     }
 
-    public void acquireSubscribeConnection(Runnable runnable) {
-        freeSubscribeConnectionsCounter.acquire(runnable);
+    public CompletableFuture<Void> acquireSubscribeConnection() {
+        return freeSubscribeConnectionsCounter.acquire();
     }
 
     public void releaseSubscribeConnection() {

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

@@ -39,7 +39,6 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.BiConsumer;
 
 /**
  * Base connection pool class 
@@ -106,10 +105,8 @@ abstract class ConnectionPool<T extends RedisConnection> {
             return;
         }
 
-        acquireConnection(entry, new Runnable() {
-            
-            @Override
-            public void run() {
+        CompletableFuture<Void> f = acquireConnection(entry, null);
+        f.thenAccept(r -> {
             CompletableFuture<T> promise = new CompletableFuture<T>();
             createConnection(entry, promise);
             promise.whenComplete((conn, e) -> {
@@ -166,12 +163,11 @@ abstract class ConnectionPool<T extends RedisConnection> {
                     }
                 }
             });
-            }
-        }, null);
+        });
     }
 
-    protected void acquireConnection(ClientConnectionsEntry entry, Runnable runnable, RedisCommand<?> command) {
-        entry.acquireConnection(runnable, command);
+    protected CompletableFuture<Void> acquireConnection(ClientConnectionsEntry entry, RedisCommand<?> command) {
+        return entry.acquireConnection(command);
     }
 
     protected abstract int getMinimumIdleSize(ClientConnectionsEntry entry);
@@ -218,17 +214,18 @@ abstract class ConnectionPool<T extends RedisConnection> {
         return acquireConnection(command, entry);
     }
 
-    public abstract static class AcquireCallback<T> implements Runnable, BiConsumer<T, Throwable> {
-        
-    }
-    
     protected final CompletableFuture<T> acquireConnection(RedisCommand<?> command, ClientConnectionsEntry entry) {
         CompletableFuture<T> result = new CompletableFuture<T>();
 
-        Runnable callback = () -> {
+        CompletableFuture<Void> f = acquireConnection(entry, command);
+        f.thenAccept(r -> {
             connectTo(entry, result, command);
-        };
-        acquireConnection(entry, callback, command);
+        });
+        result.whenComplete((r, e) -> {
+            if (e != null) {
+                f.completeExceptionally(e);
+            }
+        });
         return result;
     }
         

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

@@ -58,8 +58,8 @@ public class PubSubConnectionPool extends ConnectionPool<RedisPubSubConnection>
     }
 
     @Override
-    protected void acquireConnection(ClientConnectionsEntry entry, Runnable runnable, RedisCommand<?> command) {
-        entry.acquireSubscribeConnection(runnable);
+    protected CompletableFuture<Void> acquireConnection(ClientConnectionsEntry entry, RedisCommand<?> command) {
+        return entry.acquireSubscribeConnection();
     }
     
     @Override

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

@@ -27,22 +27,24 @@ import java.util.concurrent.atomic.AtomicInteger;
 public class AsyncSemaphore {
 
     private final AtomicInteger counter;
-    private final Queue<Runnable> listeners = new ConcurrentLinkedQueue<>();
+    private final Queue<CompletableFuture<Void>> listeners = new ConcurrentLinkedQueue<>();
 
     public AsyncSemaphore(int permits) {
         counter = new AtomicInteger(permits);
     }
     
     public boolean tryAcquire(long timeoutMillis) {
-        CountDownLatch latch = new CountDownLatch(1);
-        Runnable runnable = () -> latch.countDown();
-        acquire(runnable);
-        
+        CompletableFuture<Void> f = acquire();
         try {
-            return latch.await(timeoutMillis, TimeUnit.MILLISECONDS);
+            f.get(timeoutMillis, TimeUnit.MILLISECONDS);
+            return true;
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             return false;
+        } catch (ExecutionException e) {
+            throw new IllegalStateException(e);
+        } catch (TimeoutException e) {
+            return false;
         }
     }
 
@@ -54,23 +56,33 @@ public class AsyncSemaphore {
         listeners.clear();
     }
 
-    public void acquire(Runnable listener) {
-        listeners.add(listener);
+    public CompletableFuture<Void> acquire() {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        listeners.add(future);
         tryRun();
+        return future;
+    }
+
+    public void acquire(Runnable listener) {
+        acquire().thenAccept(r -> listener.run());
     }
 
     private void tryRun() {
+        while (true) {
             if (counter.decrementAndGet() >= 0) {
-            Runnable listener = listeners.poll();
-            if (listener == null) {
+                CompletableFuture<Void> future = listeners.poll();
+                if (future == null) {
                     counter.incrementAndGet();
                     return;
                 }
 
-            listener.run();
-        } else {
-            if (counter.incrementAndGet() > 0) {
-                tryRun();
+                if (future.complete(null)) {
+                    return;
+                }
+            }
+
+            if (counter.incrementAndGet() <= 0) {
+                return;
             }
         }
     }