Repository: netty/netty
Issue: FixedChannelPool does not count acquired channels precisely
Labels: defect

Netty version: 4.0.10.Beta5

Context:
When queued `AcquireTask` fails, `decrementAndRunTaskQueue()` method could throw assertion error.

Steps to reproduct:
1. Call `FixedChannelPool.acquire` enough to make some `AcquireTask`s be queued.
2. Close server channel abrubtly to make `AcquireTask` fail.
3. AssertionError can be produced.

``` java
assert acquiredChannelCount >= 0;
```

Opinion:
- Set a flag `acquired` in `AcquireListener` and set it true when it is dequed from `pendingAcquireQueue`.
- When acquire future is failed, if `acquired` flag is true call `decrementAndRunTaskQueue()` else call `runTaskQueue()`

``` java

    private class AcquireListener implements FutureListener<Channel> {
        private final Promise<Channel> originalPromise;
        protected boolean acquired;

        AcquireListener(Promise<Channel> originalPromise) {
            this.originalPromise = originalPromise;
            acquired = true;
        }

        @Override
        public void operationComplete(Future<Channel> future) throws Exception {
            assert executor.inEventLoop();

            if (future.isSuccess()) {
                originalPromise.setSuccess(future.getNow());
            } else {
                // Something went wrong try to run pending acquire tasks.
                if(acquired) {
                    decrementAndRunTaskQueue();
                } else {
                    runTaskQueue();
                }
                originalPromise.setFailure(future.cause());
            }
        }
    }

    private final class AcquireTask extends AcquireListener {
        final Promise<Channel> promise;
        final long expireNanoTime = System.nanoTime() + acquireTimeoutNanos;
        ScheduledFuture<?> timeoutFuture;

        public AcquireTask(Promise<Channel> promise) {
            super(promise);
            // We need to create a new promise as we need to ensure the AcquireListener runs in the correct
            // EventLoop.
            this.promise = executor.<Channel>newPromise().addListener(this);
            acquired = false;
        }
    }

    private void runTaskQueue() {
        while (acquiredChannelCount < maxConnections) {
            AcquireTask task = pendingAcquireQueue.poll();
            if (task == null) {
                break;
            }

            // Cancel the timeout if one was scheduled
            ScheduledFuture<?> timeoutFuture = task.timeoutFuture;
            if (timeoutFuture != null) {
                timeoutFuture.cancel(false);
            }

            task.acquired = true;

            --pendingAcquireCount;
            ++acquiredChannelCount;

            super.acquire(task.promise);
        }

        // We should never have a negative value.
        assert pendingAcquireCount >= 0;
        assert acquiredChannelCount >= 0;
    }

```


Issue comments:
@fratboy - Thanks for reporting!

@normanmaurer - Assigned to you for now.

@fratboy thanks... fix in the works.

@normanmaurer My pleasure.

@fratboy could you verify https://github.com/netty/netty/pull/4016 ?

@normanmaurer It's exactly same with my patch in my project. It works perfect. 

@fratboy - Thanks for the report and verification!

Why `pendingAcquireCount` use int not AtomicInteger ? @normanmaurer @fratboy 
@wyzssw because everything is done from the same Thread (EventLoop)
@normanmaurer I get it, thanks 

Pull request review comments:

Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
index 5c08ac6260a..f23c5404eb0 100644
--- a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
+++ b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
@@ -271,6 +271,8 @@ private void runTaskQueue() {
                 timeoutFuture.cancel(false);
             }
 
+            task.acquired();
+
             --pendingAcquireCount;
             ++acquiredChannelCount;
 
@@ -289,7 +291,7 @@ private final class AcquireTask extends AcquireListener {
         ScheduledFuture<?> timeoutFuture;
 
         public AcquireTask(Promise<Channel> promise) {
-            super(promise);
+            super(promise, false);
             // We need to create a new promise as we need to ensure the AcquireListener runs in the correct
             // EventLoop.
             this.promise = executor.<Channel>newPromise().addListener(this);
@@ -322,9 +324,15 @@ public final void run() {
 
     private class AcquireListener implements FutureListener<Channel> {
         private final Promise<Channel> originalPromise;
+        protected boolean acquired;
 
         AcquireListener(Promise<Channel> originalPromise) {
+            this(originalPromise, true);
+        }
+
+        protected AcquireListener(Promise<Channel> originalPromise, boolean acquired) {
             this.originalPromise = originalPromise;
+            this.acquired = acquired;
         }
 
         @Override
@@ -334,11 +342,19 @@ public void operationComplete(Future<Channel> future) throws Exception {
             if (future.isSuccess()) {
                 originalPromise.setSuccess(future.getNow());
             } else {
-                // Something went wrong try to run pending acquire tasks.
-                decrementAndRunTaskQueue();
+                if (acquired) {
+                    decrementAndRunTaskQueue();
+                } else {
+                    runTaskQueue();
+                }
+
                 originalPromise.setFailure(future.cause());
             }
         }
+
+        public void acquired() {
+            acquired = true;
+        }
     }
 
     @Override


Test patch: