Repository: netty/netty
Issue: FixedChannelPool does not properly counts the acquiredChannelCount
Labels: defect

Steps to reproduce:
1) create FixedChannelPool with maxConnections=1

``` java
FixedChannelPool(bootstrap.remoteAddress(key.host(), key.port()),
 new AbstractChannelPoolHandler() {... },1);
```

2) Request channel from FixedChannelPool
3) Make request through channel and once done release it
4) Kill the connection while channel is in pool.
5) Request channel from the pool. Since connection is not healthy anymore it will close the channel and right after that will try to acquire new channel. But acquiredChannelCount is never decremented. 
So at the point when FixedChannelPool will try to acquire connection it will think that acquiredChannelCount==1 and it won't force new connection creation, instead it will put task to the pendingAcquireQueue, see lines 195 to 226 in FixedChannelPool.

Expected result:
When connection is died(becomes unhealthy) while being in pool it should recognize that in FixedChannlePool and decrement acquiredChannelCount before asking for new channel creation.
Actual result:
acquiredChannelCount stays unchanged so new channel never gets created/acquired.

Suggested solution:
in SimpleChannelPool add code:

``` java
  private void onChannelUnhealthy(Promise<Void> channelClosePromise) {
        channelClosePromise.addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                channelClosedCauseUnhealthy();
            }
        });
    }

    protected void channelClosedCauseUnhealthy() {

    }
```

and modify notifyHealthCheck(...) to be as following:

``` java
 private void notifyHealthCheck(Future<Boolean> future, Channel ch, Promise<Channel> promise) {
        assert ch.eventLoop().inEventLoop();

        if (future.isSuccess()) {
            if (future.getNow() == Boolean.TRUE) {
                try {
                    ch.attr(POOL_KEY).set(this);
                    handler.channelAcquired(ch);
                    promise.setSuccess(ch);
                } catch (Throwable cause) {
                    Promise<Void> channelClosePromise = ch.eventLoop().<Void>newPromise();
                    onUnhealthyChannelClosed(channelClosePromise);
                    closeAndFail(ch, cause, promise,channelClosePromise);
                }
            } else {
                Promise<Void> channelClosePromise = ch.eventLoop().<Void>newPromise();
                onUnhealthyChannelClosed(channelClosePromise);
                closeChannel(ch, channelClosePromise);
                acquire(promise);
            }
        } else {
            Promise<Void> channelClosePromise = ch.eventLoop().<Void>newPromise();
            onUnhealthyChannelClosed(channelClosePromise);
            closeChannel(ch, channelClosePromise);
            acquire(promise);
        }
    }
```

closeChannel function needs to be modified as well:

``` java
 private static void closeChannel(final Channel channel, final Promise channelClosePromise) {
        channel.attr(POOL_KEY).getAndSet(null);
        ChannelFuture future = channel.close();
        if (channelClosePromise != null) {
            future.addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    channelClosePromise.setSuccess(future.isSuccess());
                }
            });
        }
    }
  private static void closeAndFail(Channel channel, Throwable cause, Promise<?> promise,
                                     Promise<Void> channelClosePromise) {
        closeChannel(channel, channelClosePromise);
        promise.setFailure(cause);
    }
```

in FixedChannelPool add method:

``` java
   protected void channelClosedCauseUnhealthy() {
       decrementAndRunTaskQueue();
    }
```


Issue comments:
Please review pull request: https://github.com/netty/netty/pull/4027

Will do tomorrow!

> Am 26.07.2015 um 20:30 schrieb ioanbsu notifications@github.com:
> 
> Please review pull request: #4027
> 
> —
> Reply to this email directly or view it on GitHub.

Applied all the changes to satisfy #4027 pull requests comments.
Unfortunately I somehow closed my original pull request. Probably because I tried to do force push so my commits are squashed.. 

I had to create new pull request that contains all my original changes and changes that were suggested by @normanmaurer  in my original pull request.
Please review https://github.com/netty/netty/pull/4029

Fixed by #4076 


Pull request review comments:
channelWasNotAcuired -> channelWasNotAcquired (typo)

final

Shouldn't we call also "channelClosePromise.setFailure(...)" if not success full ?

You can use FutureListener

You can use FutureListener

Don't we have to execute this with `FixedChannelPool.executor` ?

Hm. should I just do

``` java
channelClosePromise.setFailure(new IllegalStateException("Failed to close unhealthy channel."));
```

?

Sure.


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 f23c5404eb0..b4f7f17037b 100644
--- a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
+++ b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
@@ -245,6 +245,24 @@ public void operationComplete(Future<Void> future) throws Exception {
         return p;
     }
 
+    /**
+     * Called after channel failed to be acquired so we know to decrease the {@link #acquiredChannelCount} and pull the
+     * next pending task from {@link #pendingAcquireQueue}.
+     */
+    @Override
+    protected void channelClosedCauseUnhealthy() {
+        if (executor.inEventLoop()) {
+            decrementAndRunTaskQueue();
+        } else {
+            executor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    decrementAndRunTaskQueue();
+                }
+            });
+        }
+    }
+
     private void decrementAndRunTaskQueue() {
         --acquiredChannelCount;
 
@@ -342,9 +360,7 @@ public void operationComplete(Future<Channel> future) throws Exception {
             if (future.isSuccess()) {
                 originalPromise.setSuccess(future.getNow());
             } else {
-                if (acquired) {
-                    decrementAndRunTaskQueue();
-                } else {
+                if (!acquired) {
                     runTaskQueue();
                 }
 
diff --git a/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java b/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java
index 8526bbc5ac5..402e6335441 100644
--- a/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java
+++ b/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java
@@ -24,6 +24,7 @@
 import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.Promise;
 import io.netty.util.internal.EmptyArrays;
 import io.netty.util.internal.OneTimeTask;
@@ -50,6 +51,13 @@ public class SimpleChannelPool implements ChannelPool {
     private final ChannelPoolHandler handler;
     private final ChannelHealthChecker healthCheck;
     private final Bootstrap bootstrap;
+    private final FutureListener<Void> channelWasNotAcquired = new FutureListener<Void>() {
+        @Override
+        public void operationComplete(Future<Void> future) throws Exception {
+            channelClosedCauseUnhealthy();
+        }
+    };
+
 
     /**
      * Creates a new instance using the {@link ChannelHealthChecker#ACTIVE}.
@@ -161,18 +169,31 @@ private void notifyHealthCheck(Future<Boolean> future, Channel ch, Promise<Chann
                     handler.channelAcquired(ch);
                     promise.setSuccess(ch);
                 } catch (Throwable cause) {
-                    closeAndFail(ch, cause, promise);
+                    Promise<Void> channelClosePromise = ch.eventLoop().newPromise();
+                    channelClosePromise.addListener(channelWasNotAcquired);
+                    closeAndFail(ch, cause, promise, channelClosePromise);
                 }
             } else {
-                closeChannel(ch);
+                Promise<Void> channelClosePromise = ch.eventLoop().newPromise();
+                channelClosePromise.addListener(channelWasNotAcquired);
+                closeChannel(ch, channelClosePromise);
                 acquire(promise);
             }
         } else {
-            closeChannel(ch);
+            Promise<Void> channelClosePromise = ch.eventLoop().newPromise();
+            channelClosePromise.addListener(channelWasNotAcquired);
+            closeChannel(ch, channelClosePromise);
             acquire(promise);
         }
     }
 
+    /**
+     * Called once channel failed to be acquired.
+     * This is useful for the cases when we need to perform cleanup operations after channel failed to be acquired.
+     */
+    protected void channelClosedCauseUnhealthy() {
+    }
+
     /**
      * Bootstrap a new {@link Channel}. The default implementation uses {@link Bootstrap#connect()},
      * sub-classes may override this.
@@ -205,7 +226,7 @@ public void run() {
                 });
             }
         } catch (Throwable cause) {
-            closeAndFail(channel, cause, promise);
+            closeAndFail(channel, cause, promise, null);
         }
         return promise;
     }
@@ -218,28 +239,42 @@ private void doReleaseChannel(Channel channel, Promise<Void> promise) {
                          // Better include a stracktrace here as this is an user error.
                          new IllegalArgumentException(
                                  "Channel " + channel + " was not acquired from this ChannelPool"),
-                         promise);
+                         promise, null);
         } else {
             try {
                 if (offerChannel(channel)) {
                     handler.channelReleased(channel);
                     promise.setSuccess(null);
                 } else {
-                    closeAndFail(channel, FULL_EXCEPTION, promise);
+                    closeAndFail(channel, FULL_EXCEPTION, promise, null);
                 }
             } catch (Throwable cause) {
-                closeAndFail(channel, cause, promise);
+                closeAndFail(channel, cause, promise, null);
             }
         }
     }
 
-    private static void closeChannel(Channel channel) {
+    private static void closeChannel(final Channel channel, final Promise<Void> channelClosePromise) {
         channel.attr(POOL_KEY).getAndSet(null);
-        channel.close();
+        ChannelFuture future = channel.close();
+        if (channelClosePromise != null) {
+            future.addListener(new FutureListener<Void>(){
+
+                @Override
+                public void operationComplete(Future<Void> future) throws Exception {
+                    if (future.isSuccess()) {
+                        channelClosePromise.setSuccess(null);
+                    } else{
+                        channelClosePromise.setFailure(new IllegalStateException("Failed to close unhealthy channel."));
+                    }
+                }
+            });
+        }
     }
 
-    private static void closeAndFail(Channel channel, Throwable cause, Promise<?> promise) {
-        closeChannel(channel);
+    private static void closeAndFail(Channel channel, Throwable cause, Promise<?> promise,
+                                     Promise<Void> channelClosePromise) {
+        closeChannel(channel, channelClosePromise);
         promise.setFailure(cause);
     }
 


Test patch: